五、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。

Flink1.13架构全集| 一文带你由浅入深精通Flink方方面面（二）