Flink DataStream中CoGroup实现原理与三种 join 实现

CoGroup

CoGroup 表示联合分组，将两个不同的DataStream联合起来，在相同的窗口内按照相同的key分组处理，先通过一个demo了解其使用方式：

1. case class Order(id:String, gdsId:String, amount:Double)

2. case class Gds(id:String, name:String)

3. case class RsInfo(orderId:String, gdsId:String, amount:Double, gdsName:String)

4. object CoGroupDemo{

5. def main(args:Array[String]):Unit={

6.    val env =StreamExecutionEnvironment.getExecutionEnvironment

7.    env.setParallelism(1)

8.    val kafkaConfig =newProperties();

9.    kafkaConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

10.    kafkaConfig.put(ConsumerConfig.GROUP_ID_CONFIG,"test1");

11.    val orderConsumer =newFlinkKafkaConsumer011[String]("topic1",newSimpleStringSchema, kafkaConfig)

12.    val gdsConsumer =newFlinkKafkaConsumer011[String]("topic2",newSimpleStringSchema, kafkaConfig)

13.    val orderDs = env.addSource(orderConsumer)

14. .map(x =>{

15.        val a = x.split(",")

16. Order(a(0), a(1), a(2).toDouble)

17. })

18.    val gdsDs = env.addSource(gdsConsumer)

19. .map(x =>{

20.        val a = x.split(",")

21. Gds(a(0), a(1))

22. })

23. 
    

24.    orderDs.coGroup(gdsDs)

25. .where(_.gdsId)// orderDs 中选择key

26. .equalTo(_.id)//gdsDs中选择key

27. .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))

28. .apply(newCoGroupFunction[Order,Gds,RsInfo]{

29. overridedef coGroup(first: lang.Iterable[Order], second: lang.Iterable[Gds],out:Collector[RsInfo]):Unit={

30. //得到两个流中相同key的集合

31. }

32. })

33.    env.execute()

34. }

    }

从源码角度分析CoGrop的实现

1. 两个DataStream进行CoGroup得到的是一个CoGroupedStreams类型，后面的where、equalTo、window、apply之间的一些转换，最终得到一个WithWindow类型，包含两个dataStream、key选择、where条件、window等属性

2. 重点：WithWindow 的apply方法

• 对两个DataStream打标签进行区分，得到TaggedUnion,TaggedUnion包含one、two两个属性，分别对应两个流

• 将两个打标签后的流TaggedUnion 进行union操作合并为一个DataStream类型流unionStream

• unionStream根据不同的流选择对应where/equalTo条件进行keyBy 得到KeyedStream流

• 通过指定的window方式得到一个WindowedStream，然后apply一个被CoGroupWindowFunction包装之后的function，后续就是window的操作

到这里已经将一个CoGroup操作转换为window操作，接着看后续是如何将相同的key的两个流的数据如何组合在一起的

1.  在用户定义CoGroupFunction 被CoGroupWindowFunction包装之后，会接着被InternalIterableWindowFunction包装，一个窗口相同key的所有数据都会在一个Iterable中， 会将其传给CoGroupWindowFunction

2.  在CoGroupWindowFunction中，会将不同流的数据区分开来得到两个list,传给用户自定义的CoGroupFunction中

JOIN

在理解了coGroup的实现后，join实现原理也就比较简单，DataStream join 同样表示连接两个流，也是基于窗口实现，其内部调用了CoGroup的调用链，使用姿势p与调用流程跟CoGroup及其相似，主要有以下两点不同:

1. 不在使用CoGroupFunction，而是JoinFunction，在JoinFunction里面得到的是来自不同两个流的相同key的每一对数据

2. 函数调用链

   

   中间增加了FlatJoinCoGroupFunction函数调用，使用嵌套遍历方式得到两个流的笛卡尔积传给用户自定义函数

   
   

   

Left/Right join实现分析

Flink 中DataStream 只提供了inner join 的实现，并未提供left join 与 right join 的实现，那么同样可以通过CoGroup来实现这两种join,以left join 为例，处理逻辑在CoGroupFunction中，实现如下：

1. overridedef coGroup(first: lang.Iterable[Order], second: lang.Iterable[Gds],out:Collector[RsInfo]):Unit={

2. 
   

3.          first.foreach(x =>{

4. if(!second.isEmpty){

5.               second.foreach(y=>{

6. out.collect(newRsInfo(x.id,x.gdsId,x.amount,y.name))

7. })

8. }

9. if(second.isEmpty){

10. out.collect(newRsInfo(x.id,x.gdsId,x.amount,null))

11. }

12. })

13. }