Flink,join，coGroup，connect，union的区别、使用和实

冷不防 2021-09-24 10:48 889阅读 0赞

### join： ###

1 可用于DataStream和DataSet。只能2个DataStream一起join，或者2个DataSet一起join

2 用于DataStream时返回是JoinedStreams  ,用于DataSet时返回是JoinOperatorSets

3 用于DataStream时需要与窗口同时使用，语法是：join where equalTo window apply ，用于DataSet时的语法是：join where equalTo with （where是指定第一个输入的分区字段，equalTo是指定第二个输入的分区字段，这2个字段类型需要一致）

4 与SQL中的inner join同义，只输出2个实时窗口内或2个数据集合内能匹配上的笛卡尔积，不能匹配上的不输出。

5 apply方法中或with方法中均可以使用JoinFunction或 FlatJoinFunction处理匹配上的数据对（用于DataStream和DataSet时均可）

6 侧重对2个输入里的 数据对 进行处理，join方法的入参是单个数据

7 可以join2个类型不同的流或join2个类型不同的数据集（比如Tuple2<String, Long> join Tuple2<Long,String>）,但是匹配的key或field类型要一致，不然报错（比如where中的String与equalTo中的String匹配才行）

[Flink,join双流，join实例，java版本][Flink_join_join_java]

[Flink,join DataSet数据集，join实例，java版本][Flink_join DataSet_join_java]

### coGroup： ###

1 可用于DataStream和DataSet。只能2个DataStream一起coGroup，或者2个DataSet一起coGroup

2 用于DataStream时返回是CoGroupedStreams，用于DataSet时返回是CoGroupOperatorSets

3 用于DataStream时需要与窗口同时使用，语法是：coGroup where equalTo window apply ，用于DataSet时的语法是：coGroup where equalTo with,

4 把2个实时窗口内或2个数据集合内key相同的数据分组同一个分区，key不能匹配上的数据（只在一个窗口或集合内存在的数据）也分组到另一个分区上。

5 apply方法中或with方法中均可以使用CoGroupFunction对数据分组（用于DataStream和DataSet时均可，无FlatCoGroupFunction）

6 侧重对2个输入的 集合 进行处理，coGroup方法的入参是Iterable类型

7 可以coGroup2个类型不同的流或coGroup2个类型不同的数据集（比如Tuple2<String, Long> join Tuple2<Long,String>）,但是匹配的key或field类型要一致，不然报错（比如where中的String与equalTo中的String匹配才行）

[Flink,coGroup双流，coGroup实例，java版本][Flink_coGroup_coGroup_java]

[Flink,coGroup DataSet数据集，coGroup实例，java版本][Flink_coGroup DataSet_coGroup_java]

### connect: ###

1 只能用于DataStream,返回是ConnectedStreams。不能用于DataSet.

2 只能2个流一起connect（stream1.connect(stream2)）

3 connect后可以对2个流分别处理（使用CoMapFunction或CoFlatMapFunction）

4  可以connect2个类型不同的流（比如Tuple2<String, Long> connect Tuple2<Long,String>）

[Flink,connect双流，connect实例，java版本][Flink_connect_connect_java]

### union: ###

1 用于DataStream时,返回是Datastream;用于DataSet时,返回是DataSet;

2 可以多个流一起合并（stream1.union(stream2,stream3,stream4)），合并结果是一个新Datastream；只能2个DataSet一起合并，合并结果是一个新DataSet

3 无论是合并Datastream还是合并DataSet，都不去重，2个源的消息或记录都保存。

4 不可以union 2个类型不同的流或union 2个类型不同的数据集

[Flink,union双流，union实例，java版本][Flink_union_union_java]

[Flink,union DataSet数据集，union实例，java版本][Flink_union DataSet_union_java]

\---------------------------------更多精辟的见解可见下方----------------------------------

![format_png][]

[https://www.jianshu.com/p/aa7d0f6d0dc4][https_www.jianshu.com_p_aa7d0f6d0dc4]

1 ：

![format_png 1][]

## ConnectedStreams ##

在 DataStream 上有一个 union 的转换 `dataStream.union(otherStream1, otherStream2, ...)`，用来合并多个流，新的流会包含所有流中的数据。union 有一个限制，就是所有合并的流的类型必须是一致的。`ConnectedStreams` 提供了和 union 类似的功能，用来连接**两个**流，但是与 union 转换有以下几个区别：

1.  ConnectedStreams 只能连接两个流，而 union 可以连接多于两个流。
2.  ConnectedStreams 连接的两个流类型可以不一致，而 union 连接的流的类型必须一致。
3.  ConnectedStreams 会对两个流的数据应用不同的处理方法，并且双流之间可以共享状态。这在第一个流的输入会影响第二个流时, 会非常有用。

如下 ConnectedStreams 的样例，连接 `input` 和 `other` 流，并在`input`流上应用`map1`方法，在`other`上应用`map2`方法，双流可以共享状态（比如计数）。

<table> 
 <tbody> 
  <tr> 
   <td> <pre><code>val input: DataStream[MyType] = ...
val other: DataStream[AnotherType] = ...
 
val connected: ConnectedStreams[MyType, AnotherType] = input.connect(other)
 
val result: DataStream[ResultType] = 
        connected.map(new CoMapFunction[MyType, AnotherType, ResultType]() {
            override def map1(value: MyType): ResultType = { ... }
            override def map2(value: AnotherType): ResultType = { ... }
        })
</code></pre> </td> 
  </tr> 
 </tbody> 
</table>

当并行度为2时，其执行图如下所示：

[![format_png 2][]][format_png 2 1]

[http://wuchong.me/blog/2016/05/20/flink-internals-streams-and-operations-on-streams/][http_wuchong.me_blog_2016_05_20_flink-internals-streams-and-operations-on-streams]

2：

## union ##

在`DataStream`上使用`union`算子可以合并多个同类型的数据流，并生成同类型的数据流，即可以将多个`DataStream[T]`合并为一个新的`DataStream[T]`。数据将按照先进先出（First In First Out）的模式合并，且不去重。下图`union`对白色和深色两个数据流进行合并，生成一个数据流。

![union示意图][union]

假设股票价格数据流来自不同的交易所，我们将其合并成一个数据流：

val shenzhenStockStream: DataStream[StockPrice] = ...
    val hongkongStockStream: DataStream[StockPrice] = ...
    val shanghaiStockStream: DataStream[StockPrice] = ...
    val unionStockStream: DataStream[StockPrice] = shenzhenStockStream.union(hongkongStockStream, shanghaiStockStream)
    复制代码

## connect ##

`union`虽然可以合并多个数据流，但有一个限制，即多个数据流的数据类型必须相同。`connect`提供了和`union`类似的功能，用来连接两个数据流，它与`union`的区别在于：

1.  `connect`只能连接两个数据流，`union`可以连接多个数据流。
2.  `connect`所连接的两个数据流的数据类型可以不一致，`union`所连接的两个数据流的数据类型必须一致。
3.  两个`DataStream`经过`connect`之后被转化为`ConnectedStreams`，`ConnectedStreams`会对两个流的数据应用不同的处理方法，且双流之间可以共享状态。

`connect`经常被应用在对一个数据流使用另外一个流进行控制处理的场景上，如下图所示。控制流可以是阈值、规则、机器学习模型或其他参数。

![对一个数据流进行控制处理][format_png 3]

对于`ConnectedStreams`，我们需要重写`CoMapFunction`或`CoFlatMapFunction`。这两个接口都提供了三个泛型，这三个泛型分别对应第一个输入流的数据类型、第二个输入流的数据类型和输出流的数据类型。在重写函数时，对于`CoMapFunction`，`map1`处理第一个流的数据，`map2`处理第二个流的数据；对于`CoFlatMapFunction`，`flatMap1`处理第一个流的数据，`flatMap2`处理第二个流的数据。Flink并不能保证两个函数调用顺序，两个函数的调用依赖于两个数据流数据的流入先后顺序，即第一个数据流有数据到达时，`map1`或`flatMap1`会被调用，第二个数据流有数据到达时，`map2`或`flatMap2`会被调用。

[https://juejin.im/post/5e037bf4e51d4557f638c4ed][https_juejin.im_post_5e037bf4e51d4557f638c4ed]

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5LqM5Y2B5YWt55S755Sf55qE5Y2a5a6i_size_20_color_FFFFFF_t_70_g_se_x_16][]

[Flink_join_join_java]: https://mp.csdn.net/console/editor/html/106783608
[Flink_join DataSet_join_java]: https://mp.csdn.net/console/editor/html/106803782
[Flink_coGroup_coGroup_java]: https://mp.csdn.net/console/editor/html/106796006
[Flink_coGroup DataSet_coGroup_java]: https://mp.csdn.net/console/editor/html/106802756
[Flink_connect_connect_java]: https://mp.csdn.net/console/editor/html/106802294
[Flink_union_union_java]: https://mp.csdn.net/console/editor/html/106804745
[Flink_union DataSet_union_java]: https://mp.csdn.net/console/editor/html/106804869
[format_png]: /images/20210923/0fd5a91a96844246806dc8c4add2d0c2.png
[https_www.jianshu.com_p_aa7d0f6d0dc4]: https://www.jianshu.com/p/aa7d0f6d0dc4
[format_png 1]: /images/20210923/5ced2e06fe6640b5a2d841d032fd83dc.png
[format_png 2]: /images/20210923/051e702f20d64b0ea2c92e3b4ec3d773.png
[format_png 2 1]: http://img3.tbcdn.cn/5476e8b07b923/TB1WEjuJVXXXXcHXFXXXXXXXXXX
[http_wuchong.me_blog_2016_05_20_flink-internals-streams-and-operations-on-streams]: http://wuchong.me/blog/2016/05/20/flink-internals-streams-and-operations-on-streams/
[union]: /images/20210923/a773698eea22470b88a3a16380f4c86b.png
[format_png 3]: /images/20210923/44177c412a5c45139ac6f9c884b66611.png
[https_juejin.im_post_5e037bf4e51d4557f638c4ed]: https://juejin.im/post/5e037bf4e51d4557f638c4ed
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5LqM5Y2B5YWt55S755Sf55qE5Y2a5a6i_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20210923/f32fc729bb8042a28ca16efbd102b29d.png