转载-Flink Sql on Zeppelin——双流Join

朴灿烈づ我的快乐病毒、 2022-12-05 15:21 180阅读 0赞

## 双流Join ##

### 概述 ###

*  其实之前和大家聊过双流Join，这次之所以再拉出来讲并不是在炒冷饭，而是发现了之前讲的一些不足
 *  以UnBounded Join来说，我之前一直以为，无论哪边的流先到，只要匹配上之后，key对应的、存储在state中的数据应该被清理。这样，无论哪边再来一条同样key的数据，不会触发数据下发，也就是一对一。其实这样的理解是错误的，key会一直存储在state中，直到OOM或者配置TTL之后被清理。那么，也就是会多次触发数据下发，也就是会发生一对多、多对多
 *  之前关于Time Interval Join的理解，也有一点问题。官网并没有写`LFET JOIN`的语法，我就也以为只支持`INNER JOIN`，责任在于我，没有自己去测试。同样的，也会有上面UnBounded Join的特性：一对多、多对多
 *  简单的补全了一点不足之处，下面让我们来通过案例，来补全我们对双流Join的认知把

### UnBounded Join ###

*  UnBounded Join这边，以上次讲解的为基础，再看一下新的东西，之前没看过双流Join的小伙伴可以[点我查看][Link 1]，之前的内容就不再赘述了
 *  这次主要给大家演示一下，一对多，多对多的情况下，会产生的结果
 *  先启动一下两个任务
    
        %flink.ssql(type=update)
        
        select * from unBounded_join1 t1 inner join unBounded_join2 t2 on t1.order_id = t2.order_id
    
        %flink.ssql(type=update)
        
        select * from unBounded_join1 t1 left join unBounded_join2 t2 on t1.order_id = t2.order_id
 *  再插入这样的数据到kafka
    
        %flink.ssql
        
        insert into unBounded_join1 select 1,1;
        insert into unBounded_join1 select 1,1;
        insert into unBounded_join2 select 1,1001;
        -- insert into unBounded_join2 select 1,1002;
 *  我们观察一下刚才的任务结果  
    ![1][]
 *  果然，会是一个一对多的结果。那么此时，我们再插入一条同样的key的数据到右表，看看会是怎么样的（将被注释的那条语句取消注释然后执行；别的不执行）  
    ![2][]
 *  实际上我们只在左表插入了两条数据，但结果却出现了四条数据，这就是因为key并没有被删除，而是继续缓存，和我之前的理解有误。这样带来的火锅就是state无限增大，直到OOM

### Time Interval Join ###

*  和UnBounded Join一样，Time Interval Join也会有一对多、多对多的特点。优点是key对应的数据，在超出时间范围之后，将会被删除
 *  我们来聊一下Time Interval Join在不同的时间类型下，`INNER JOIN`和`LEFT JOIN`的不同表现
 *  让我们启动四个任务
    
        %flink.ssql(type=update)
        
        select t1.user_id,t1.order_id,t1.ts,t2.order_id,t2.item_id,t2.ts from timeInterval_join1 t1 
        inner join timeInterval_join2 t2 
        on t1.order_id = t2.order_id 
        and (t2.r_t between t1.r_t - interval '10' second and t1.r_t + interval '10' second )
    
        %flink.ssql(type=update)
        
        select t1.user_id,t1.order_id,t1.ts,t2.order_id,t2.item_id,t2.ts from timeInterval_join1 t1 
        left join timeInterval_join2 t2 
        on t1.order_id = t2.order_id 
        and (t2.r_t between t1.r_t - interval '10' second and t1.r_t + interval '10' second )
    
        %flink.ssql(type=update)
        
        select t1.user_id,t1.order_id,t1.ts,t2.order_id,t2.item_id,t2.ts from timeInterval_join1 t1 
        inner join timeInterval_join2 t2 
        on t1.order_id = t2.order_id 
        and (t2.p between t1.p - interval '10' second and t1.p + interval '10' second )
    
        %flink.ssql(type=update)
        
        select t1.user_id,t1.order_id,t1.ts,t2.order_id,t2.item_id,t2.ts from timeInterval_join1 t1 
        left join timeInterval_join2 t2 
        on t1.order_id = t2.order_id 
        and (t2.p between t1.p - interval '10' second and t1.p + interval '10' second )
 *  然后让我们插入数据。注意一点的是，第一条语句先执行，执行完过10秒以上，再执行第二条语句
    
        %flink.ssql
        
        -- 先执行我，执行完过10秒以上再执行下面被注释的语句
        insert into timeInterval_join1 select 1,1,'2020-06-28 14:21:12';
        
        -- 我得过10秒再执行
        -- insert into timeInterval_join2 select 1,1001,'2020-06-28 14:21:12';
 *  直接看最终结果了  
    ![3][]
 *  用event time的 join，是能够正常输出数据的，因为他们的event time在规定的范围内
 *  而process time的join，`INNER JOIN`没有数据输出，因为超过时间范围；`LEFT JOIN`会输出`NULL`，因为已经超过了时间约束（过了10秒以上才插入数据到右表），都符合我们的语义
 *  这些就是我上次漏讲的一些内容，最后给大家留个小作业，如果是基于event time的`LEFT JOIN`，超时了会输出什么呢？大家可以自行测试

### 写在最后 ###

*  首先，感谢一下微信好友`一杯咖啡`，是他指点出我之前文章的不足，感谢！
 *  其次，也为自己的不严谨给大家道个歉，差点误人子弟；同样，如果别的文章或者说这篇也有不足，希望大家都给我指出来告诉我，不胜感谢！
 *  再者，我最近都比较忙，所以文章更新比较慢，量也比较少，大家多多谅解，事情太多了，下班都得干活
 *  最后，如果大家想直接用我的note，可以去我的[github][]里面下载

[Link 1]: https://blog.csdn.net/weixin_47482194/article/details/106317517
[1]: /images/20221123/0192d3b6aa594a8b93ca1d297946c9c6.png
[2]: /images/20221123/a76760df4314451681af6586c701f3d9.png
[3]: /images/20221123/3e1b8285632e41f2b87ccd30efabbde3.png
[github]: https://github.com/lonelyGhostisdog/flinksql/tree/master/src/main/resources/Flink%20on%20Zeppelin