五、Flink 和其他框架对比

“

下面比较Spark和Flink的不同。一些方法在两个框架中都是相同的，而有些方法有很大不同。

”

5.1 Flink vs Spark5.2 三大实时计算框架整体对比

“

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官网。

另外需要特别说明的是：

课程中全部程序采用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, stdoutlog4j.appender.stdout=org.apache.log4j.ConsoleAppenderlog4j.appender.stdout.layout=org.apache.log4j.PatternLayoutlog4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

“

接下来我们用一个最简单的示例来说明Flink代码怎样编写：统计一段文字中，每个单词出现的频次。这就是传说中的WordCount程序。

源码位于src/main/java目录下。首先新建一个包，命名为com.liuhao，在这个包下我们将编写Flink入门的WordCount程序。

”

6.3.1 批处理

对于批处理而言，输入的应该是收集好的数据集。这里我们可以将要统计的文字，写入一个文本文档，然后读取这个文件处理数据就可以了。

在工程根目录下新建一个input文件夹，并在下面创建文本文件words.txt在words.txt中输入一些文字，例如：

hello flinkhello worldhello sparkhello hadoophello hive

我们进行单词频次统计的基本思路是：先逐行读入文件数据，然后将每一行文字拆分成单词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次。

具体代码实现如下：

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

“

我们会发现程序启动之后没有任何输出、也不会退出。这是正常的——因为Flink的流处理是事件驱动的，当前程序会一直处于监听状态，只有接收到数据才会执行任务、输出统计结果。

”

从hadoop01发送数据。

“

我们会发现，输出的结果与之前读取文件的流处理非常相似。而且可以非常明显地看到，每输入一条数据，就有一次对应的输出。具体对应关系是：输入“hello flink”，就会输出两条统计结果（flink，1）和（hello，1）；之后再输入“hello world”，同样会将hello和world的个数统计输出，hello的个数会对应增长为2。