2.4 数据流API_Flink内核原理与实现-QQ阅读男生中文武侠网

上QQ阅读APP看书，第一时间看更新

2.4 数据流API

DataStream API是Flink流计算应用中最常用的API，相比Table & SQL API更加底层、更加灵活。

2.4.1 数据读取

数据读取的API定义在StreamExecutionEnvironment，这是Flink流计算应用的起点，第一个DataStream就是从数据读取API中构造出来的。在Flink中，除了内置的数据读取API外，还针对不同类型的外部存储系统提供了对应的Connector连接器，使用连接器也能够实现数据读取的目的。

1.从内存读取数据

Flink提供了一系列的方法，直接在内存中生成数据，方便测试和演示。API如图2-4所示。

图2-4 内存数据读取API

2.文件读取数据

内置的从文件中读取数据的API如图2-5所示。

图2-5 读取文件API

从文件中读取分为读取文本文件和一般文件两类，文本文件无须多说，一般文件指的是带有结构的文件，如Avro、Parquet等。

文件读取的模式有一次性读取FileProcessingMode.PROCESS_ONCE和持续读取FileProcessingMode.PROCESS_CONTINUOUSLY。如果不指定则默认为一次性读取。使用持续读取模式时，可以设定读取间隔，单位为ms。间隔越小实时性越高，资源消耗相应变多，反之则实时性越低，资源消耗降低。

3. Socket接入数据

Socket接入数据即从网络端口接收数据。内置的从Socket接入数据的API如图2-6所示。

图2-6 Socket接入数据API

socketTextStream（）的参数比较简单，需要提供hostname（主机名）、port（端口号）、delimiter（分隔符）和maxRetry（最大重试次数）。

4.自定义读取

自定义数据读取就是使用Flink连接器、自定义数据读取函数，与外部存储交互，读取数据，如从Kafka、JDBC、HDFS等读取。自定义数据读取的API如图2-7所示。

图2-7 自定义数据读取API

addSource（）方法本质上来说依赖于Flink的SourceFunction体系，与外部的存储进行交互。createInput（）方法底层调用的是addSource（）方法，封装为InputFormatSourceFunction，所以自定义读取方式的本质就是实现自定义的SourceFunction。关于SourceFunction，将在第3章进行详细介绍。

2.4.2 处理数据

DataStream API使用Fluent风格处理数据，在开发的时候其实是在编写一个DataStream转换过程，形成了DataStream处理链，在Flink开发章节有过阐述。调用DataStream API生成新的DataStream的转换关系如图2-8所示。

图2-8 DataStream相互转换关系

从图中可以看到，并不是所有的DataStream都可以相互转换。

1. Map

接收1个元素，输出1个元素。Map应用在DataStream上，输出结果为DataStream。

DataStream#map运算对应的是MapFunction，其类泛型为MapFunction<T,O>，T代表输入数据类型（Map方法的参数类型），O代表操作结果输出类型（Map方法返回的数据类型），如代码清单2-1所示。

代码清单2-1 Map代码示例

2. FlatMap

接收1个元素，输出0、1、…、N个元素。该类运算应用在DataStream上，输出结果为DataStream。

DataStream#flatMap接口对应的是FlatMapFunction，其类泛型为FlatMapFunction<T,O>，T代表输入数据类型（FlatMap方法的参数类型），O代表操作结果输出类型，如代码清单2-2所示。

代码清单2-2 FlatMap接口示例

3. Filter

过滤数据，如果返回true则该元素继续向下传递，如果为false则将该元素过滤掉。该类运算应用在DataStream上，输出结果为DataStream。

DataStream#filter接口对应的是FilterFunction，其类泛型为FilterFunction<T>，T代表输入和输出元素的数据类型，如代码清单2-3所示。

代码清单2-3 Filter代码示例

4. KeyBy

将数据流元素进行逻辑上的分组，具有相同Key的记录将被划分到同一分组。KeyBy（）使用Hash Partitioner实现。该运算应用在DataStream上，输出结果为KeyedStream。

输出的数据流的类型为KeyedStream<T,KEY>，其中T代表KeyedStream中元素数据类型，KEY代表逻辑Key的数据类型，如代码清单2-4所示。

代码清单2-4 KeyBy代码示例

以下两种数据不能作为Key。

1）POJO类未重写hashCode（），使用了默认的Object.hashCode（）。

2）数组类型。

5. Reduce

按照KeyedStream中的逻辑分组，将当前数据与最后一次的Reduce结果进行合并，合并逻辑由开发者自己实现。该类运算应用在KeyedStream上，输出结果为DataStream。

ReduceFunction<T>中的T代表KeyedStream中元素的数据类型，如代码清单2-5所示。

代码清单2-5 Reduce代码示例

6. Fold

Fold与Reduce类似，区别在于Fold是一个提供了初始值的Reduce，用初始值进行合并运算。该类运算应用在KeyedStream上，输出结果为DataStream。

Folder接口对应的是FoldFunction，其类泛型为FoldFunction<O, T>，O为KeyStream中的数据类型，T为初始值类型和Fold方法返回值类型，如代码清单2-6所示。

代码清单2-6 Fold代码示例

FoldFunction<O, T>已经被标记为Deprecated废弃，替代接口是AggregateFunction<IN, ACC, OUT>。

7. Aggregation

渐进聚合具有相同Key的数据流元素，以min和minBy为例，min返回的是整个KeyedStream的最小值，minBy按照Key进行分组，返回每个分组的最小值。在KeyedStream上应用聚合运算输出结果为DataStream，如代码清单2-7所示。

代码清单2-7 内置聚合运算代码示例

8. Window

对KeyedStream的数据，按照Key进行时间窗口切分，如每5秒钟一个滚动窗口，每个key都有自己的窗口。该类运算应用在KeyedStream上，输出结果为WindowedStream。

输出结果的类泛型为WindowedStream<T, K, W extends Window>，T为KeyedStream中的元素数据类型，K为指定Key的数据类型，W为窗口类型，如代码清单2-8所示。

代码清单2-8 Window代码示例

关于窗口，第4章会有详细讲解。

9. WindowAll

对一般的DataStream进行时间窗口切分，即全局1个窗口，如每5秒钟一个滚动窗口。应用在DataStream上，输出结果为AllWindowedStream，如代码清单2-9所示。

代码清单2-9 WindowAll代码示例

注意：在一般的DataStream上进行窗口切分，往往会导致无法并行计算，所有的数据会集中到WindowAll算子的一个Task上。

关于窗口请参照Window原理和机制章节。

10. Window Apply

将Window函数应用到窗口上，Window函数将一个窗口的数据作为整体进行处理。Window Stream有两种：分组后的WindowedStream和未分组的AllWindowedStream。

（1）WindowedStream

在WindowedStream上应用的是WindowFunction，在WindowStream应用此类运算，输出结果为DataStream。WindowFunction<IN, OUT, KEY, W extends Window>中的IN表示输入值的类型，OUT表示输出值的类型，KEY表示Key的类型，W表示窗口的类型，如代码清单2-10所示。

代码清单2-10 WindowFunction代码示例

（2）AllWindowedStream

在AllWindowedStream上应用的是AllWindowFunction，输出结果为DataStream。该类运算对应的是AllWindowFunction，其类泛型定义为AllWindowFunction<IN, OUT, W extends Window>，IN表示输入值的类型，OUT表示输出值的类型，W表示窗口的类型，如代码清单2-11所示。

代码清单2-11 AllWindowFunction代码示例

11. Window Reduce

在WindowedStream上应用ReduceFunction，输出结果为DataStream。参见前面的Reduce章节，如代码清单2-12所示。

代码清单2-12 Window Reduce代码示例

12. Window Fold

在WindowedStream上应用FoldFunction，输出结果为DataStream，参见前面的Fold章节，如代码清单2-13所示。

代码清单2-13 Window Fold代码示例

13. Window Aggregation

统计聚合运算，在WindowedStream应用该运算，输出结果为DataStream。

在WindowedStream上应用AggregationFunction，参见前面的Aggregations章节，如代码清单2-14所示。

代码清单2-14 内置的Window聚合运算代码示例

14. Union

把两个或多个DataStream合并，所有DataStream中的元素都会组合成一个新的DataStream，但是不去重。如果在自身上应用Union运算，则每个元素在新的DataStream出现两次，如代码清单2-15所示。

代码清单2-15 Union运算示例

15. Window Join

在相同时间范围的窗口上Join两个DataStream数据流，输出结果为DataStream。

Join核心逻辑在JoinFunction<IN1,IN2,OUT>中实现，IN1为第一个DataStream中的数据类型，IN2为第二个DataStream中的数据类型，OUT为Join结果的数据类型，如代码清单2-16所示。

代码清单2-16 Join代码示例

16. Interval Join

对两个KeyedStream进行Join，需要指定时间范围和Join时使用的Key，输出结果为DataStream。

例如对于事件e1和e2，Key相同，时间判断条件为：

Join的核心逻辑在ProcessJoinFunction<IN1,IN2,OUT>中实现，IN1为第一个DataStream中元素数据类型，IN2为第二个DataStream中的元素数据类型，OUT为结果输出类型，如代码清单2-17所示。

代码清单2-17 Interval Join代码示例

17. WindowCoGroup

两个DataStream在相同时间窗口上应用CoGroup运算，输出结果为DataStream，CoGroup和Join功能类似，但是更加灵活。

CoGroup接口对应的是CoGroupFunction，其类泛型为CoGroupFunction<IN1, IN2, O>，IN1代表第一个DataStream中的元素类型，IN2代表第二个DataStream中的元素类型，O为输出结果类型，如代码清单2-18所示。

代码清单2-18 CoGroup代码示例

18. Connect

连接（connect）两个DataStream输入流，并且保留其类型，输出流为ConnectedStream。两个数据流之间可以共享状态。

输出数据流的类泛型为ConnectedStreams<IN1,IN2>，IN1代表第1个数据流中的数据类型，IN2表示第2个数据流中的数据类型，如代码清单2-19所示。

代码清单2-19 Connect代码示例

19. CoMap和CoFlatMap

在ConnectedStream上应用Map和FlatMap运算，输出流为DataStream。其基本逻辑类似于在一般DataStream上的Map和FlatMap运算，区别在于CoMap转换有2个输入，Map转换有1个输入，CoFlatMap同理，如代码清单2-20所示。

代码清单2-20 CoMap和CoFlatMap代码示例

20. Split

将DataStream按照条件切分为多个DataStream，输出流为SplitDataStream。该方法已经标记为Deprecated废弃，推荐使用SideOutput，如代码清单2-21所示。

代码清单2-21 Split代码示例

21. Select

Select与Split运算配合使用，在Split运算中切分的多个DataStream中，Select用来选择其中某一个具体的DataStream，如代码清单2-22所示。

代码清单2-22 Select代码示例

22. Iterate

在API层面上，对DataStream应用迭代会生成1个IteractiveStream，然后在IteractiveStream上应用业务处理逻辑，最终生成1个新的DataStream，IteractiveStream本质上来说是一种中间数据流对象。

在数据流中创建一个迭代循环，即将下游的输出发送给上游重新处理。如果一个算法会持续地更新模型，这种情况下反馈循环比较有用，如代码清单2-23所示。

代码清单2-23 Iterate代码示例

23. Extract Timestamps

从记录中提取时间戳，并生成Watermark。该类运算不会改变DataStream，如代码清单2-24所示。

代码清单2-24 提取时间戳代码示例

24. Project

该类运算只适用于Tuple类型的DataStream，使用Project选取子Tuple，可以选择Tuple的部分元素，可以改变元素顺序，类似于SQL语句中的Select子句，输出流仍然是DataStream，如代码清单2-25所示。

代码清单2-25 Project代码示例

2.4.3 数据写出

数据读取的API绑定在StreamExecutionEnvironment上，数据写出的API绑定在DataStream对象上。在现在的版本中，只有写到Console控制台、Socket网络端口、自定义三类，写入文本文件、CSV文件等文件接口都已被标记为废弃了。接口使用的详细介绍参照官方文档即可。

自定义数据写出接口是DataStream.addSink，对于Sink的详细介绍参见连接器和输出函数章节。

2.4.4 旁路输出

旁路输出在Flink中叫作SideOutput，用途类似于DataStream#split，本质上是一个数据流的切分行为，按照条件将DataStream切分为多个子数据流，子数据流叫作旁路输出数据流，每个旁路输出数据流可以有自己的下游处理逻辑。如图2-9所示，通过旁路输出将正常和异常的数据分别记录到不同的外部存储中。