五、Flink 和其他框架对比
“下面比较Spark和Flink的不同。一些方法在两个框架中都是相同的，而有些方法有很大不同。
”
5.1 Flink vs Spark
| Spark Streaming | Flink |
| DStream | DataStream |
| Trasnformation | Trasnformation |
| Action | Sink |
| Task | SubTask |
| Pipeline | Oprator chains |
| DAG | DataFlow Graph |
| Master + Driver | JobManager |
| Worker + Executor | TaskManager |
5.2 三大实时计算框架整体对比
| 框架 | 优点 | 缺点 |
| Storm | 低延迟 | 吞吐量低、不能保证 exactly-once、编程 API 不 丰富 |
| Spark Streaming | 吞吐量高、可以保证 exactly-once、编程 API 丰富 | 延迟较高 |
| Flink | 低延迟、吞吐量高、可以保证 exactly-once、编程 API 丰富 | 快速迭代中,API 变化比较快 |
“Spark 就是为离线计算而设计的，在 Spark 生态体系中，不论是流处理和批处理都是底层引 擎都是 Spark Core，Spark Streaming 将微批次小任务不停的提交到 Spark 引擎，从而实现准 实时计算，SparkStreaming 只不过是一种特殊的批处理而已。
”
“Flink 就是为实时计算而设计的，Flink 可以同时实现批处理和流处理，Flink 将批处理（即有 有界数据）视作一种特殊的流处理。
”
5.3 数据处理架构
“从根本上说，Spark和Flink采用了完全不同的数据处理方式。可以说，两者的世界观是截然不同的。
Spark以批处理为根本，并尝试在批处理之上支持流计算；在Spark的世界观中，万物皆批次，离线数据是一个大批次，而实时数据则是由一个一个无限的小批次组成的。所以对于流处理框架Spark Streaming而言，其实并不是真正意义上的“流”处理，而是“微批次”（micro-batching）处理。
”
“而Flink则认为，流处理才是最基本的操作，批处理也可以统一为流处理。在Flink的世界观中，万物皆流，实时数据是标准的、没有界限的流，而离线数据则是有界限的流。
”
“正因为这种架构上的不同，Spark和Flink在不同的应用领域上表现会有差别。一般来说， Spark 基于微批处理的方式做同步总有一个“攒批”的过程，所以会有额外开销，因此无法在流处理的低延迟上做到极致。在低延迟流处理场景，Flink 已经有明显的优势。而在海量数据的批处理领域，Spark能够处理的吞吐量更大，加上其完善的生态和成熟易用的API，目前同样优势比较明显。
”
5.4 数据模型和运行架构
“Spark底层数据模型是弹性分布式数据集（RDD），Spark Streaming 进行微批处理的底层接口DStream，实际上处理的也是一组组小批数据RDD的集合。
而Flink的基本数据模型是数据流（DataFlow），以及事件（Event）序列。
数据模型不同，对应在运行处理的流程上，自然也会有不同的架构。Spark做批计算，需要将任务对应的DAG划分阶段（Stage），一个完成后经过shuffle再进行下一阶段的计算。而Flink是标准的流式执行模式，一个事件在一个节点处理完后可以直接发往下一个节点进行处理。
”
5.5 Spark还是Flink呢
“Spark和Flink可以说目前是各擅胜场，批处理领域Spark称王，而在流处理方面Flink当仁不让。具体到项目应用中，不仅要看是流处理还是批处理，还需要在延迟、吞吐量、可靠性，以及开发容易度等多个方面进行权衡。
如果在工作中需要从Spark和Flink这两个主流框架中选择一个来进行实时流处理，我们更加推荐使用Flink，主要的原因有：
- Flink的延迟是毫秒级别，而Spark Streaming的延迟是秒级延迟。
- Flink提供了严格的精确一次性语义保证。
- Flink的窗口API更加灵活、语义更丰富。
- Flink提供事件时间语义，可以正确处理延迟数据。
- Flink提供了更加灵活的对状态编程的API。
当然，在海量数据的批处理方面，Spark还是具有明显的优势。而且Spark的生态更加成熟，也会使其在应用中更为方便。相信随着Flink的快速发展和完善，这方面的差距会越来越小。
另外，这两大框架也在不停地互相借鉴、取长补短。Spark 2.0之后新增的Structured Streaming流处理引擎借鉴DataFlow进行了大量优化，同样做到了低延迟、时间正确性以及精确一次性语义保证；Spark 2.3以后引入的连续处理（Continuous Processing）模式，更是可以在至少一次语义保证下做到1毫秒的延迟。而Flink自1.9版本合并Blink以来，在SQL的表达和批处理的能力上同样有了长足的进步。
”
六、Flink直接上手
6.1 准备环境
“在进行代码的编写之前，先将我们使用的开发环境和工具介绍一下：
系统环境为Windows 10。
需提前安装Java 8。
集成开发环境（IDE）使用IntelliJ IDEA，具体的安装流程参见IntelliJ官网。https://www.jetbrains.com/zh-cn/idea/download/#section=windows
另外需要特别说明的是：
课程中全部程序采用Java语言编写；
课程中使用的Flink版本为1.13.0。
”
6.2 创建项目
“在准备好所有的开发环境之后，我们就可以开始开发自己的第一个Flink程序了。首先我们要做的，就是在IDEA中搭建一个Flink项目的骨架。我们会使用Java项目中常见的Maven来进行依赖管理。
”
1. 创建工程
- 打开IntelliJ IDEA，创建一个Maven工程。
- 将这个Maven工程命名为FlinkTutorial。
- 将这个Maven工程命名为FlinkTutorial。
2. 添加项目依赖
“在项目的pom文件中，增加标签设置属性，然后增加标签引入需要的依赖。我们需要添加的依赖最重要的就是Flink的相关组件，包括flink-java、flink-streaming-java，以及flink-clients（客户端，也可以省略）。另外，为了方便查看运行日志，我们引入slf4j和log4j进行日志管理。
”
<properties> <flink.version>1.13.0</flink.version> <java.version>1.8</java.version> <scala.binary.version>2.12</scala.binary.version> <slf4j.version>1.7.30</slf4j.version> </properties> <dependencies> <!-- 引入Flink相关依赖--> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <!-- 引入日志管理相关依赖--> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>${slf4j.version}</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>${slf4j.version}</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-to-slf4j</artifactId> <version>2.14.0</version> </dependency> </dependencies>
“这里做一点解释：在属性中，我们定义了<scala.binary.version>，这指代的是所依赖的Scala版本。这有一点奇怪：Flink底层是Java，而且我们也只用Java API，为什么还会依赖Scala呢？这是因为Flink的架构中使用了Akka来实现底层的分布式通信，而Akka是用Scala开发的。我们本书中用到的Scala版本为2.12。
”
3. 配置日志管理
在目录src/main/resources下添加文件:log4j.properties，内容配置如下：
log4j.rootLogger=error, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
6.3 编写代码
“接下来我们用一个最简单的示例来说明Flink代码怎样编写：统计一段文字中，每个单词出现的频次。这就是传说中的WordCount程序。
源码位于src/main/java目录下。首先新建一个包，命名为com.liuhao，在这个包下我们将编写Flink入门的WordCount程序。
”
6.3.1 批处理
对于批处理而言，输入的应该是收集好的数据集。这里我们可以将要统计的文字，写入一个文本文档，然后读取这个文件处理数据就可以了。
- 在工程根目录下新建一个input文件夹，并在下面创建文本文件words.txt
- 在words.txt中输入一些文字，例如：
hello flink hello world hello spark hello hadoop hello hive
- 在com.liuhao包下新建Java类wordcountP，在静态main方法中编写测试代码。
我们进行单词频次统计的基本思路是：先逐行读入文件数据，然后将每一行文字拆分成单词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次。
具体代码实现如下：
package com.liuhao.wordcountP; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.operators.AggregateOperator; import org.apache.flink.api.java.operators.DataSource; import org.apache.flink.api.java.operators.FlatMapOperator; import org.apache.flink.api.java.operators.UnsortedGrouping; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.util.Collector; publicclass wordcountP { public static void main(String[] args) throws Exception { // 1.获取执行环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); // 2.读取文件数据 DataSource<String> DataSource = env.readTextFile("input/wordcount.txt"); // 3.处理转换，变成（hello，1）二元组 FlatMapOperator<String, Tuple2<String, Long>> stringTuple2FlatMapOperator = DataSource .flatMap((String line, Collector<Tuple2<String, Long>> out) -> { String[] word = line.split(" "); for (String s : word) { out.collect(Tuple2.of(s, 1l)); } }).returns(Types.TUPLE(Types.STRING, Types.LONG)); // 4.按照key进行分组 UnsortedGrouping<Tuple2<String, Long>> groupBy = stringTuple2FlatMapOperator.groupBy(0); // 5.按照第二个字段进行sum AggregateOperator<Tuple2<String, Long>> sum = groupBy.sum(1); // 6.输出 sum.print(); } }
需要注意的是，这种代码的实现方式，是基于DataSet API的，也就是我们对数据的处理转换，是看作数据集来进行操作的。事实上Flink本身是流批统一的处理架构，批量的数据集本质上也是流，没有必要用两套不同的API来实现。所以从Flink 1.12开始，官方推荐的做法是直接使用DataStream API，在提交任务时通过将执行模式设为BATCH来进行批处理：
$ bin/flink run -Dexecution.runtime-mode=BATCH WordCount.jar
这样，DataSet API就没什么用了，在实际应用中我们只要维护一套DataStream API就可以。这里只是为了方便大家理解，我们依然用DataSet API做了批处理的实现。
6.3.2 流处理
对于Flink而言，流才是整个处理逻辑的底层核心，所以流批统一之后的DataStream API更加强大，可以直接处理批处理和流处理的所有场景。
我们就针对不同类型的输入数据源，用具体的代码来实现流处理。
1.读取文件
我们同样试图读取文档words.txt中的数据，并统计每个单词出现的频次。整体思路与之前的批处理非常类似，代码模式也基本一致。
在com.liuhao包下新建Java类WordCountL，在静态main方法中编写测试代码。具体代码实现如下：
package com.liuhao.WordCountL; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; import java.util.Arrays; publicclass WordCountL { public static void main(String[] args) throws Exception { // 1. 创建流式执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取文件 DataStreamSource<String> lineStream = env.readTextFile("input/words.txt"); // 3. 转换、分组、求和，得到统计结果 SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineStream.flatMap((String line, Collector<Tuple2<String, Long>> out) -> { String[] words = line.split(" "); for (String word : words) { out.collect(Tuple2.of(word, 1L)); } }).returns(Types.TUPLE(Types.STRING, Types.LONG)) .keyBy(data -> data.f0) .sum(1); // 4. 打印 sum.print(); // 5. 执行 env.execute(); } }
主要观察与批处理程序WordCountP的不同：
“创建执行环境的不同，流处理程序使用的是StreamExecutionEnvironment。
转换处理之后，得到的数据对象类型不同。
分组操作调用的是keyBy方法，可以传入一个匿名函数作为键选择器（KeySelector），指定当前分组的key是什么。
代码末尾需要调用env的execute方法，开始执行任务。
”
2.读取文本流
在实际的生产环境中，真正的数据流其实是无界的，有开始却没有结束，这就要求我们需要保持一个监听事件的状态，持续地处理捕获的数据。
为了模拟这种场景，我们就不再通过读取文件来获取数据了，而是监听数据发送端主机的指定端口，统计发送来的文本数据中出现过的单词的个数。具体实现上，我们只要对WordCountL代码中读取数据的步骤稍做修改，就可以实现对真正无界流的处理。
- 将WordCountL代码中读取文件数据的readTextFile方法，替换成读取socket文本流的方法socketStreamTextStream。具体代码实现如下：
package com.liuhao.SocketWordCountL ; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; import java.util.Arrays; publicclass SocketWordCountL { public static void main(String[] args) throws Exception { // 1. 创建流式执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取文本流 DataStreamSource<String> lineStream = env.socketTextStream("hadoop102", 7777); // 3. 转换、分组、求和，得到统计结果 SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineStream.flatMap((String line, Collector<Tuple2<String, Long>> out) -> { String[] words = line.split(" "); for (String word : words) { out.collect(Tuple2.of(word, 1L)); } }).returns(Types.TUPLE(Types.STRING, Types.LONG)) .keyBy(data -> data.f0) .sum(1); // 4. 打印 result.print(); // 5. 执行 env.execute(); } }
代码说明和注意事项
socket文本流的读取需要配置两个参数：发送端主机名和端口号。这里代码中指定了主机“hadoop102”的7777端口作为发送数据的socket端口，读者可以根据测试环境自行配置。
在实际项目应用中，主机名和端口号这类信息往往可以通过配置文件，或者传入程序运行参数的方式来指定。
socket文本流数据的发送，可以通过Linux系统自带的netcat工具进行模拟。
- 在Linux环境的主机hadoop102上，执行下列命令，发送数据进行测试：
[hadoop@hadoop ~]$ nc -lk 7777
- 启动WordCountL程序
“我们会发现程序启动之后没有任何输出、也不会退出。这是正常的——因为Flink的流处理是事件驱动的，当前程序会一直处于监听状态，只有接收到数据才会执行任务、输出统计结果。
”
- 从hadoop01发送数据。
“我们会发现，输出的结果与之前读取文件的流处理非常相似。而且可以非常明显地看到，每输入一条数据，就有一次对应的输出。具体对应关系是：输入“hello flink”，就会输出两条统计结果（flink，1）和（hello，1）；之后再输入“hello world”，同样会将hello和world的个数统计输出，hello的个数会对应增长为2。
