Flink之数据流类型

Flink为流处理和批处理分别提供了DataStream API和DataSet API。在开发工作中这些API极大的便利了开发者开发大数据应用。DataStream 在经过Transformation之后，会随之生成相应的数据流类型。比如：KeyedStream，JoinedStreams，CoGroupedStreams等等。但这些数据流类型之间是如何通过转换联系在一起的？。下文☞☞☞☞☞

流之间转换

图1

DataStream

DataStream 是Flink 流处理API中最核心的数据结构。他代表了一个运行在多个分区上的并行流。一个DataStream可以从StreamExecutionEnvironment 通过env.addSource(SourceFunction)获得，可以但是不限于这一种方式。

DataStream上的转换操作都是逐条的，比如map(),flatMap(),filter()。DataStream 也可以执行rebalance(再平衡，用来减轻数据倾斜) 和broadcaseted(广播)等分区转换。

val stream: DataStream[MyType] = env.addSource(new FlinkKafkaConsumer011[String](...)).setParallelism(3)
val str1: DataStream[(String, MyType)] = stream.flatMap { ... }
val str2: DataStream[(String, MyType)] = stream.rebalance()
val str3: DataStream[AnotherType] = stream.map { ... }

上面给出的DataStream代码块在运行时会转换成如下执行图：

图2

如上图的执行图所示，DataStream各个算子会并行运行，算子之间是数据流分区。如Source的第一个并行实例（S1）和flatMap()的第一个并行实例（f1）之间就是一个数据流分区。而flatMap()和map()之间犹豫加了rebalance()，它们之间的数据流分区就有3个自分区（f1的数据流向map()实例）。这与Apache Kafka是很类似的。把流想象成Kafka Topic，而一个流分区就表示一个Topic Partition,流的目标并行算子实例就是Kafka Consumers。

KeyedStream

KeyedStream用来表示根据指定的key进行分组的数据流。一个KeyStream可以通过调用DataStream.keyBy()来获得。而再KeyedStream上进行任何transformation都将转变回DataStream。而实现中，KeyStream是把key的信息写入到了transformation中。每条记录只能访问所属key的状态，其上的聚合函数可以方便地操作和保存对应key的状态。

WindowedStream & AllWindowedStream

WindowedStream代表了根据key分组，并且基于WindowAssigner分割窗口的数据流。所以WindowedStream都是从KeyStream衍生而来。而在WindowedStream上进行任何transformation也都将转变回DataStream。

val stream: DataStream[MyType] = ...
val windowed: WindowedDataStream[MyType] = stream
        .keyBy("userId")
        .window(TumblingEventTimeWindows.of(Time.seconds(5))) // Last 5 seconds of data
val result: DataStream[ResultType] = windowed.reduce(myReducer)

上述WindowedStream的样例代码在运行时会转换成如下的执行图：

图3

Flink的窗口实现中会将到达的数据缓存再对应的窗口buffer中（一个数据可能会对应多个窗口）。当到达窗口发送的条件时（由Trigger控制），Flink会对整个窗口中的数据进行处理。Flink在聚合类窗口有一定的优化，即不会保存窗口中的所有值，而是得到一个元素执行一次聚合函数，最终只会保存一份数据即可。

在key分组的流上进行窗口切分是比较常用的场景，也能够很好地并行化（不同的key上的窗口聚合可以分配到不同的task上去处理）。不过有时候我们也需要再普通流上进行窗口的操作，这就是AllWindowedStream。AllWindowedStream是直接在DataStream上进行windowAll(...)操作。AllWindowedStream的实现是基于WindowedStream的。Flink不推荐使用AllWindowedStream，因为在普通流上进行窗口操作，就势必需要将所有的分区的流都汇集到单个task中，而这个单个的Task很显然就会成为整个Job的瓶颈。

JoinedStream & CoGroupedStreams

双流Join也是一个非常常见的。在源码中可以发现JoinedStream和CoGroupedStreams的代码实现大部分都是一样的，JoinedStreams在底层又调用了CoGroupedStream来实现Join功能。除了名字不一样，一开始很难将他们区分开，而且为什么要提供两个功能类似的接口呢？

实际上着两者还是有区别的，首先co-group侧重的是group,是对同一个key上的两组集合进行操作，而join侧重的是pair,是对同一个key上的每对元素操作。co-group比join更通用一些，因为join只是co-group的一个特例，所以join是可以基于co-group来实现的（当然有优化的空间）。而在co-group之外又提供了join接口是因为用户更熟悉join，而且能够跟DataSet API保持一致，降低用户的学习成本。

JoinedStreams和CoGroupedStreams是基于Window上实现的，所以CoGroupStreams最终又调用了WindowedStream来实现。

val firstInput: DataStream[MyType] = ...
val secondInput: DataStream[AnotherType] = ...
 
val result: DataStream[(MyType, AnotherType)] = firstInput.join(secondInput)
    .where("userId").equalTo("id")
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new JoinFunction () {...})

上述JoinedStreams的样例代码在运行时会转变成如下的执行图：

图4

双流上的数据在同一个key会被分别分配到同一个Window窗口的左右两个篮子里，当window结束的时候，会对左右篮子进行笛卡尔积从而得到每一对pair,对每一对pair应用JoinFunction。目前join窗口的双流数据都是被缓存在内存中的，也就是说如果某个key上的窗口数据太多就会导致JVM OOM(然而数据倾斜是常态)。双流join的难点也正是这里，这也是社区对join操作的优化方向，列入可以借鉴Flink在批处理join中的优化方案，也可以用ManagedMemory来管理窗口中的数据，并当数据超过阀值时能spill到磁盘上。

ConnectedStreams

在DataStream上有一个union的转换 dataStream.union(otherStream1,otherStream2,...),用来合并多个流，新的流会包含所有流中的数据。union有一个限制，就是所有合并的流的类型必须是一致的。ConnectedStreams提供了和union类似的功能，用来连接两个流，但是与union转换有以下几个区别：
①.ConnectedStreams只能连接两个流，而union可以多余两个流；
②.ConnectedStreams连接的两个流类型可以不一致，而union连接的流的类型必须一致；
③.ConnectedStreams会对两个流的数据应用不同的处理方法，并且双流之间可以共享状态。这再第一个流的输入会影响第二流时，会非常有用。

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

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 = { ... }
        })

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

图5