Flink DataStream 编程入门

流处理是 Flink 的核心，流处理的数据集用 DataStream 表示。数据流从可以从各种各样的数据源中创建（消息队列、Socket 和文件等），经过 DataStream 的各种 transform 操作，最终输出文件或者标准输出。这个过程跟之前文章中介绍的 Flink 程序基本骨架一样。本篇介绍 DataStream 相关的入门知识。

Flink 101

为了学习 Flink 的朋友能查看到每个例子的源码，我创建了一个 GitHub 项目：github.com/duma-repo/a… 这里会存放每一篇文章比较重要的示例的源码，目前支持 Java 和 Scala，仍在不断完善中。代码下载后可以在本地运行，也可以打包放在集群上运行。同时，欢迎各位将优质的资源提交到项目中。

简单示例

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;public class WindowWordCount {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<Tuple2<String, Integer>> dataStream = env.socketTextStream("localhost", 9999).flatMap(new Splitter()).keyBy(0).timeWindow(Time.seconds(5)).sum(1);dataStream.print();env.execute("Window WordCount");}public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {@Overridepublic void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {for (String word: sentence.split(" ")) {out.collect(new Tuple2<String, Integer>(word, 1)); //空格分割后，每个单词转换成 (word, 1) 二元组输出}}}}复制代码

这个例子跟之间介绍 WordCount 的例子类似，这里详细介绍下涉及的 API 和含义

数据源：socketTextStream 是从 socket 创建的数据流，可以使用 nc -l 9000 创建 socket 客户端发送数据
transform：flatMap 将输入的数据按照空格分割后，扁平化处理（flat即为扁平的意思）；keyBy 会按照指定的 key 进行分组，这里就是将单词作为 key；timeWindow 指定时间窗口，这里是 5s 处理一次；sum 是聚合函数，将分组好的单词个数求和
输出：print 将处理完的数据输出到标准输出流中，可以在控制台看到输出的结果。调用 execute 方法提交 Job

Data Source

经过以上的介绍，我们知道常见的数据源有 socket、消息队列和文件等。对于常见的数据源 Flink 已经定义好了读取函数，接下来一一介绍。

基于文件

readTextFile(path)：读文本文件，默认是文件类型是 TextInputFormat，并且返回类型是 String
readFile(fileInputFormat, path)：读文件，需要指定输入文件的格式
readFile(fileInputFormat, path, watchType, interval, typeInfo)：以上两个方法内部都会调用这个方法，参数说明：
- fileInputFormat - 输入文件的类型
- path - 输入文件路径
- watchType - 取值为 FileProcessingMode.PROCESS_CONTINUOUSLY 和 FileProcessingMode.PROCESS_ONCE
  - FileProcessingMode.PROCESS_CONTINUOUSLY - 当输入路径下有文件被修改，整个路径下内容将会被重新处理
  - FileProcessingMode.PROCESS_ONCE - 只扫描一次，便退出。因此这种模式下输入数据只读取一次
- interval - 依赖 watchType 参数，对于 FileProcessingMode.PROCESS_CONTINUOUSLY 每隔固定时间（单位：毫秒）检测路径下是否有新数据
- typeInfo - 返回数据的类型

需要注意，在底层 Flink 将读文件的过程分为两个子任务 —— 文件监控和数据读取（reader）。监控任务由 1 个 task 实现，而读取的任务由多个 task 实现，数量与 Job 的并行度相同。监控任务的作用是扫描输入路径（周期性或者只扫描一次，取决于 watchType），当数据可以被处理时，会将数据分割成多个分片，将分片分配给下游的 reader 。一个分片只会被一个 reader 读取，一个 reader 可以读取多个分片。

基于 Socket

socketTextStream：从 socket 数据流中读数据

基于 Collection

fromCollection(Collection)：从 Java.util.Collection 类型的数据中创建输入流，collection 中的所有元素类型必须相同
fromCollection(Iterator, Class)：从 iterator (迭代器）中创建输入流，Class 参数指定从 iterator 中的数据类型
fromElements(T ...)：从给定的参数中创建输入流，所有参数类型必须相同
fromParallelCollection(SplittableIterator, Class)：从 iterator 中创建并行的输入流，Class 指定 iterator 中的数据类型
generateSequence(from, to)：从 from 至 to 之间的数据序列创建并行的数据流

自定义

addSource：可以自定义输入源，通过实现 SourceFunction 接口来自定义非并行的输入流；也可以实现 ParallelSourceFunction 接口或集成 RichParallelSourceFunction 类来自定义并行输入流，当然也可以定义好的数据源，如：Kafka，addSource(new FlinkKafkaConsumer08<>(...))

DataStream 的 transform

之前已经介绍了一些 transfrom 函数，如：map、flatMap 和 filter 等。同时还有窗口函数：window、timeWindow 等，聚合函数：sum、reduce 等。更多的 transform 函数以及使用将会单独写一篇文章介绍。

Data Sink

Data Sink 便是数据的输出。同 Data Source 类似， Flink 也内置了一些输出函数，如下：

writeAsText(path) / TextOutputFormat：将数据作为 String 类型输出到指定文件
writeAsCsv(...) / CsvOutputFormat：将 Tuple 类型输出到 ',' 分隔的 csv 类型的文件。行和列的分隔符可以通过参数配置，默认的为 '\n' 和 ','
print() / printToErr()：将数据打印到标准输出流或者标准错误流，可以指定打印的前缀。
writeUsingOutputFormat() / FileOutputFormat：输出到 OutputFormat 类型指定的文件，支持对象到字节的转换。
writeToSocket：根据 SerializationSchema 将数据输出到 socket
addSink：自定义输出函数，如：自定义将数据输出到 Kafka