书名：Flink入门与实战
ISBN：978-7-115-51678-7
本书由人民邮电出版社发行数字版。版权所有，侵权必究。
您购买的人民邮电出版社电子书仅供您个人使用，未经授权，不得以任何方式复制和传播本书内容。
我们愿意相信读者具有这样的良知和觉悟，与我们共同保护知识产权。
如果购买者有侵权行为，我们可能对该用户实施包括但不限于关闭该帐号等维权措施，并可能追究法律责任。
著 徐 葳
责任编辑 陈聪聪
人民邮电出版社出版发行 北京市丰台区成寿寺路11号
邮编 100164 电子邮件 315@ptpress.com.cn
网址 http://www.ptpress.com.cn
读者服务热线：(010)81055410
反盗版热线：(010)81055315
本书旨在帮助读者从零开始快速掌握Flink的基本原理与核心功能。本书首先介绍了Flink的基本原理和安装部署，并对Flink中的一些核心API进行了详细分析。然后配套对应的案例分析，分别使用Java代码和Scala代码实现案例。最后通过两个项目演示了Flink在实际工作中的一些应用场景，帮助读者快速掌握Flink开发。
学习本书需要大家具备一些大数据的基础知识，比如Hadoop、Kafka、Redis、Elasticsearch等框架的基本安装和使用。本书也适合对大数据实时计算感兴趣的读者阅读。
Flink项目是大数据计算领域冉冉升起的一颗新星。大数据计算引擎的发展经历了几个过程，从第1代的MapReduce，到第2代基于有向无环图的Tez，第3代基于内存计算的Spark，再到第4代的Flink。因为Flink可以基于Hadoop进行开发和使用，所以Flink并不会取代Hadoop，而是和Hadoop紧密结合。
Flink主要包括DataStream API、DataSet API、Table API、SQL、Graph API和FlinkML等。现在Flink也有自己的生态圈，涉及离线数据处理、实时数据处理、SQL操作、图计算和机器学习库等。
本书共分11章，每章的主要内容如下。
第1～2章主要针对Flink的原理组件进行分析，其中包括针对Storm、Spark Streaming和Flink这3个实时计算框架进行对比和分析，以及快速实现Flink的入门案例开发。
第3章主要介绍Flink的安装部署，包含Flink的几种部署模式：本地模式、Standalone模式和YARN模式。本章主要针对YARN模式进行了详细分析，因为在实际工作中以YARN模式为主，这样可以充分利用现有集群资源。
第4章主要针对DataStream和DataSet中的相关组件及API进行详细分析，并对Table API和SQL操作进行了基本的分析。
第5～9章主要针对Flink的一些高级特性进行了详细的分析，包含Broadcast、Accumulator、Distributed Cache、State、CheckPoint、StateBackend、SavePoint、Window、Time、Watermark以及Flink中的并行度。
第10章主要介绍常用组件Kafka-Connector，针对Kafka Consumer和Kafka Producer的使用结合具体案例进行分析，并描述了Kafka的容错机制的应用。
第11章介绍Flink在实际工作中的两个应用场景：一个是实时数据清洗(实时ETL)，另一个是实时数据报表，通过这两个项目实战可以加深对Flink的理解。
感谢所有在本书编写过程中提出宝贵意见的朋友。作者水平有限，书中如有不足之处还望指出并反馈至邮箱xuwei@xuwei.tech，作者将不胜感激。
本书由异步社区出品，社区（https://www.epubit.com/）为您提供相关资源和后续服务。
本书提供如下资源：
本书配套资源请到异步社区本书购买页处下载。
要获得以上配套资源，请在异步社区本书页面中单击
，跳转到下载界面，按提示进行操作即可。注意：为保证购书读者的权益，该操作会给出相关提示，要求输入提取码进行验证。
作者和编辑尽最大努力来确保书中内容的准确性，但难免会存在疏漏。欢迎您将发现的问题反馈给我们，帮助我们提升图书的质量。
当您发现错误时，请登录异步社区，按书名搜索，进入本书页面，单击“提交勘误”，输入勘误信息，单击“提交”按钮即可（见下图）。本书的作者和编辑会对您提交的勘误进行审核，确认并接受后，您将获赠异步社区的100积分。积分可用于在异步社区兑换优惠券、样书或奖品。
我们的联系邮箱是contact@epubit.com.cn。
如果您对本书有任何疑问或建议，请您发邮件给我们，并请在邮件标题中注明本书书名，以便我们更高效地做出反馈。
如果您有兴趣出版图书、录制教学视频，或者参与图书翻译、技术审校等工作，可以发邮件给我们；有意出版图书的作者也可以到异步社区在线提交投稿（直接访问www.epubit.com/selfpublish/submission即可）。
如果所在的学校、培训机构或企业想批量购买本书，或异步社区出版的其他图书，也可以发邮件给我们。
如果您在网上发现有针对异步社区出品图书的各种形式的盗版行为，包括对图书全部或部分内容的非授权传播，请您将怀疑有侵权行为的链接发邮件给我们。您的这一举动是对作者权益的保护，也是我们持续为您提供有价值的内容的动力之源。
“异步社区”是人民邮电出版社旗下IT专业图书社区，致力于出版精品IT技术图书和相关学习产品，为作译者提供优质出版服务。异步社区创办于2015年8月，提供大量精品IT技术图书和电子书，以及高品质技术文章和视频课程。更多详情请访问异步社区官网https://www.epubit.com。
“异步图书”是由异步社区编辑团队策划出版的精品IT专业图书的品牌，依托于人民邮电出版社近30年的计算机图书出版积累和专业编辑团队，相关图书在封面上印有异步图书的LOGO。异步图书的出版领域包括软件开发、大数据、 AI、测试、前端、网络技术等。
异步社区
微信服务号
本章讲解Flink的基本原理，主要包含Flink原理及架构分析、Flink组件介绍、Flink中的流处理和批处理的对比、Flink的一些典型应用场景分析，以及Flink和其他流式计算框架的区别等。
很多人是在2015年才听到Flink这个词的，其实早在2008年，Flink的前身就已经是柏林理工大学的一个研究性项目，在2014年这个项目被Apache孵化器所接受后，Flink迅速成为ASF（Apache Software Foundation）的顶级项目之一。截至目前，Flink的版本经过了多次更新，本书基于1.6版本写作。
Flink是一个开源的流处理框架，它具有以下特点。
Flink主要由Java代码实现，它同时支持实时流处理和批处理。对于Flink而言，作为一个流处理框架，批数据只是流数据的一个极限特例而已。此外，Flink还支持迭代计算、内存管理和程序优化，这是它的原生特性。
由图1.1可知，Flink的功能特性如下。
图1.1 Flink的功能特性
在这里解释一下，高吞吐表示单位时间内可以处理的数据量很大，低延迟表示数据产生以后可以在很短的时间内对其进行处理，也就是Flink可以支持快速地处理海量数据。
Flink架构可以分为4层，包括Deploy层、Core层、API层和Library层，如图1.2所示。
从图1.2可知， Flink对底层的一些操作进行了封装，为用户提供了DataStream API和DataSet API。使用这些API可以很方便地完成一些流数据处理任务和批数据处理 任务。
图1.2 Flink架构
读者应该对Hadoop和Storm程序有所了解，在Hadoop中实现一个MapReduce需要两个阶段——Map和Reduce，而在Storm中实现一个Topology则需要Spout和Bolt组件。因此，如果我们想实现一个Flink任务的话，也需要有类似的逻辑。
Flink中提供了3个组件，包括DataSource、Transformation和DataSink。
因此，想要组装一个Flink Job，至少需要这3个组件。
Flink Job=DataSource+Transformation+DataSink
在大数据处理领域，批处理与流处理一般被认为是两种截然不同的任务，一个大数据框架一般会被设计为只能处理其中一种任务。比如，Storm只支持流处理任务，而MapReduce、Spark只支持批处理任务。Spark Streaming是Apache Spark之上支持流处理任务的子系统，这看似是一个特例，其实不然——Spark Streaming采用了一种Micro-Batch架构，即把输入的数据流切分成细粒度的Batch，并为每一个Batch数据提交一个批处理的Spark任务，所以Spark Streaming本质上还是基于Spark批处理系统对流式数据进行处理，和Storm等完全流式的数据处理方式完全不同。
通过灵活的执行引擎，Flink能够同时支持批处理任务与流处理任务。在执行引擎层级，流处理系统与批处理系统最大的不同在于节点间的数据传输方式。
如图1.3所示，对于一个流处理系统，其节点间数据传输的标准模型是，在处理完成一条数据后，将其序列化到缓存中，并立刻通过网络传输到下一个节点，由下一个节点继续处理。而对于一个批处理系统，其节点间数据传输的标准模型是，在处理完成一条数据后，将其序列化到缓存中，当缓存写满时，就持久化到本地硬盘上；在所有数据都被处理完成后，才开始将其通过网络传输到下一个节点。
图1.3 Flink的3种数据传输模型
这两种数据传输模式是两个极端，对应的是流处理系统对低延迟和批处理系统对高吞吐的要求。Flink的执行引擎采用了一种十分灵活的方式，同时支持了这两种数据传输模型。
Flink以固定的缓存块为单位进行网络数据传输，用户可以通过设置缓存块超时值指定缓存块的传输时机。如果缓存块的超时值为0，则Flink的数据传输方式类似于前面所提到的流处理系统的标准模型，此时系统可以获得最低的处理延迟；如果缓存块的超时值为无限大，则Flink的数据传输方式类似于前面所提到的批处理系统的标准模型，此时系统可以获得最高的吞吐量。
缓存块的超时值也可以设置为0到无限大之间的任意值，缓存块的超时阈值越小，Flink流处理执行引擎的数据处理延迟就越低，但吞吐量也会降低，反之亦然。通过调整缓存块的超时阈值，用户可根据需求灵活地权衡系统延迟和吞吐量。
Flink主要应用于流式数据分析场景，目前涉及如下领域。
Flink在如下类型的公司中有具体的应用。
Storm是比较早的流式计算框架，后来又出现了Spark Streaming和Trident，现在又出现了Flink这种优秀的实时计算框架，那么这几种计算框架到底有什么区别呢？下面我们来详细分析一下，如表1.1所示。
表1.1 流式计算框架对比
| 产品 | 模型 | API | 保证次数 | 容错机制 | 状态管理 | 延时 | 吞吐量 |
|---|---|---|---|---|---|---|---|
| Storm | Native（数据进入立即处理） | 组合式（基础API） | At-least-once （至少一次） | Record ACK（ACK机制） | 无 | 低 | 低 |
| Trident | Micro-Batching（划分为小批 处理） | 组合式 | Exactly-once （仅一次） | Record ACK | 基于操作（每次操作有一个状态） | 中等 | 中等 |
| Spark Streaming | Micro-Batching | 声明式（提供封装后的高阶函数，如count函数） | Exactly-once | RDD CheckPoint（基于RDD做CheckPoint） | 基于DStream | 中等 | 高 |
| Flink | Native | 声明式 | Exactly-once | CheckPoint（Flink的一种快照） | 基于操作 | 低 | 高 |
在这里对这几种框架进行对比。
官网中Flink和Storm的吞吐量对比如图1.4所示。
图1.4 Flink和Storm的吞吐量对比
前面我们分析了3种实时计算框架，那么公司在实际操作时到底选择哪种技术框架呢？下面我们来分析一下。
第1章针对Flink的基本原理、架构和组件进行了分析，本章开始快速实现一个Flink的入门案例，这样可以加深对之前内容的理解。
在实战之前，需要先说明一下开发工具的问题。官方建议使用IntelliJ IDEA，因为它默认集成了Scala和Maven环境，使用更加方便，当然使用Eclipse也是可以的。
开发Flink程序时，可以使用Java或者Scala语言，个人建议使用Scala，因为使用Scala实现函数式编程会比较简洁。当然使用Java也可以，只不过实现起来代码逻辑比较笨重罢了。
在开发Flink程序的时候，建议使用Maven管理依赖。针对Maven仓库，建议使用国内镜像仓库地址，因为国外仓库下载较慢，可以使用国内阿里云的Maven仓库。
注意：如果发现依赖国内源无法下载的时候，记得切换回国外源。利用国内阿里云Maven仓库镜像进行相关配置时，需要修改$Maven_HOME/conf/settings.xml文件。
<mirror>
<id>aliMaven</id>
<name>aliyun Maven</name>
<url>http://Maven.aliyun.com/nexus/content/groups/public/</url>
<mirrorOf>central</mirrorOf>
</mirror>在使用Maven管理Flink程序相关依赖的时候，需要提前将它们配置好。对应的Maven项目创建完成以后，也需要在这个项目的pom.xml文件中进行相关配置。
使用Java语言开发Flink程序的时候需要添加以下配置。
注意：在这里使用的Flink版本是1.6.1。如果使用的是其他版本，需要到Maven仓库中查找对应版本的Maven配置。
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.6.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>1.6.1</version>
<scope>provided</scope>
</dependency>使用Scala语言开发Flink程序的时候需要添加下面的配置。
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId>
<version>1.6.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>1.6.1</version>
<scope>provided</scope>
</dependency>注意：在IDEA等开发工具中运行代码的时候，需要把依赖配置中的scope属性注释掉。在编译打JAR包的时候，需要开启scope属性，这样最终的JAR包就不会把这些依赖包也包含进去，因为集群中本身是有Flink的相关依赖的。
开发Flink程序有固定的流程。
（1）获得一个执行环境。
（2）加载/创建初始化数据。
（3）指定操作数据的Transaction算子。
（4）指定计算好的数据的存放位置。
（5）调用execute()触发执行程序。
注意：Flink程序是延迟计算的，只有最后调用execute()方法的时候才会真正触发执行程序。
延迟计算的好处：你可以开发复杂的程序，Flink会将这个复杂的程序转成一个Plan，并将Plan作为一个整体单元执行！
在这里，提前创建一个Flink的Maven项目，起名为FlinkExample，效果如图2.1所示。
图2.1 项目目录
后面的Java代码全部存放在src/main/Java目录下，Scala代码全部存放在src/main/Scala目录下，流计算相关的代码存放在对应的streaming目录下，批处理相关的代码则存放在对应的batch目录下。
需求分析：通过Socket手工实时产生一些单词，使用Flink实时接收数据，对指定时间窗口内（如2s）的数据进行聚合统计，并且把时间窗口内计算的结果打印出来。
首先添加Java代码对应的Maven依赖，参考2.1.2节的内容。注意，在下面的代码中，我们会创建一个WordWithCount类，这个类主要是为了方便统计每个单词出现的总次数。
需求：实现每隔1s对最近2s内的数据进行汇总计算。
分析：通过Socket模拟产生单词，使用Flink程序对数据进行汇总计算。
代码实现如下。
package xuwei.tech.streaming;
import org.apache.Flink.api.common.functions.FlatMapFunction;
import org.apache.Flink.api.Java.utils.ParameterTool;
import org.apache.Flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.Flink.runtime.state.filesystem.FsStateBackend;
import org.apache.Flink.runtime.state.memory.MemoryStateBackend;
import org.apache.Flink.streaming.api.DataStream.DataStream;
import org.apache.Flink.streaming.api.DataStream.DataStreamSource;
import org.apache.Flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.Flink.streaming.api.windowing.time.Time;
import org.apache.Flink.util.Collector;
/**
* 单词计数之滑动窗口计算
*
* Created by xuwei.tech
*/
public class SocketWindowWordCountJava {
public static void main(String[] args) throws Exception{
//获取需要的端口号
int port;
try {
ParameterTool parameterTool = ParameterTool.fromArgs(args);
port = parameterTool.getInt("port");
}catch (Exception e){
System.err.println("No port set. use default port 9000--Java");
port = 9000;
}
//获取Flink的运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
String hostname = "hadoop100";
String delimiter = "\n";
//连接Socket获取输入的数据
DataStreamSource<String> text = env.socketTextStream(hostname, port, delimiter);
// a a c
// a 1
// a 1
// c 1
DataStream<WordWithCount> windowCounts = text.flatMap(new FlatMapFunction
<String, WordWithCount>() {
public void flatMap(String value, Collector<WordWithCount> out) throws
Exception {
String[] splits = value.split("\\s");
for (String word : splits) {
out.collect(new WordWithCount(word, 1L));
}
}
}).keyBy("word")
.timeWindow(Time.seconds(2), Time.seconds(1))//指定时间窗口大小为2s，指定时间间隔为1s
.sum("count");//在这里使用sum或者reduce都可以
/*.reduce(new ReduceFunction<WordWithCount>() {
public WordWithCount reduce(WordWithCount a,
WordWithCount b) throws Exception {
return new WordWithCount(a.word,a.count+b.count);
}
})*/
//把数据打印到控制台并且设置并行度
windowCounts.print().setParallelism(1);
//这一行代码一定要实现，否则程序不执行
env.execute("Socket window count");
}
public static class WordWithCount{
public String word;
public long count;
public WordWithCount(){}
public WordWithCount(String word,long count){
this.word = word;
this.count = count;
}
@Override
public String toString() {
return "WordWithCount{" +
"word='" + word + '\'' +
", count=" + count +
'}';
}
}
}首先添加Scala代码对应的Maven依赖，参考2.1.2节的内容。在这里通过case class的方式在Scala中创建一个类。
需求：实现每隔1s对最近2s内的数据进行汇总计算。
分析：通过Socket模拟产生单词，使用Flink程序对数据进行汇总计算。
代码实现如下。
package xuwei.tech.streaming
import org.apache.Flink.api.Java.utils.ParameterTool
import org.apache.Flink.streaming.api.Scala.StreamExecutionEnvironment
import org.apache.Flink.streaming.api.windowing.time.Time
/**
* 单词计数之滑动窗口计算
*
* Created by xuwei.tech
*/
object SocketWindowWordCountScala {
def main(args: Array[String]): Unit = {
//获取Socket端口号
val port: Int = try {
ParameterTool.fromArgs(args).getInt("port")
}catch {
case e: Exception => {
System.err.println("No port set. use default port 9000--Scala")
}
9000
}
//获取运行环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//连接Socket获取输入数据
val text = env.socketTextStream("hadoop100",port,'\n')
//解析数据（把数据打平），分组，窗口计算，并且聚合求sum
//注意：必须要添加这一行隐式转行，否则下面的FlatMap方法执行会报错
import org.apache.Flink.api.Scala._
val windowCounts = text.flatMap(line => line.split("\\s"))//打平，把每一行单词都切开
.map(w => WordWithCount(w,1))//把单词转成word , 1这种形式
.keyBy("word")//分组
.timeWindow(Time.seconds(2),Time.seconds(1))//指定窗口大小，指定间隔时间
.sum("count");// sum或者reduce都可以
//.reduce((a,b)=>WordWithCount(a.word,a.count+b.count))
//打印到控制台
windowCounts.print().setParallelism(1);
//执行任务
env.execute("Socket window count");
}
case class WordWithCount(word: String,count: Long)
}在前面的案例代码中指定hostname为hadoop100，port默认为9000，表示流处理程序默认监听这个主机的9000端口。因此在执行程序之前，需要先在hadoop100这个节点上面监听这个端口，通过执行下面命令实现。
[root@hadoop100 soft]# nc -l 9000
a
b
a然后在IDEA中运行编写完成的程序代码，结果如下。
WordWithCount{word='a', count=1}
WordWithCount{word='b', count=1}
WordWithCount{word='a', count=2}
WordWithCount{word='b', count=1}
WordWithCount{word='a', count=1}前面使用Flink实现了一个典型的流式计算案例，下面来看一下Flink的另一个应用场景——Batch离线批处理。
需求：统计一个文件中的单词出现的总次数，并且把结果存储到文件中。
Java代码实现如下。
package xuwei.tech.batch;
import org.apache.Flink.api.common.functions.FlatMapFunction;
import org.apache.Flink.api.Java.DataSet;
import org.apache.Flink.api.Java.ExecutionEnvironment;
import org.apache.Flink.api.Java.operators.DataSource;
import org.apache.Flink.api.Java.tuple.Tuple2;
import org.apache.Flink.util.Collector;
/**
*单词计数之离线计算
*
* Created by xuwei.tech
*/
public class BatchWordCountJava {
public static void main(String[] args) throws Exception{
String inputPath = "D:\\data\\file";
String outPath = "D:\\data\\result";
//获取运行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//获取文件中的内容
DataSource<String> text = env.readTextFile(inputPath);
DataSet<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).groupBy(0).sum(1);
counts.writeAsCsv(outPath,"\n"," ").setParallelism(1);
env.execute("batch word count");
}
public static class Tokenizer implements FlatMapFunction<String,Tuple2<String,
Integer>>{
public void flatMap(String value, Collector<Tuple2<String, Integer>> out)
throws Exception {
String[] tokens = value.toLowerCase().split("\\W+");
for (String token: tokens) {
if(token.length()>0){
out.collect(new Tuple2<String, Integer>(token,1));
}
}
}
}
}需求：统计一个文件中的单词出现的总次数，并且把结果存储到文件中。
Scala代码实现如下。
package xuwei.tech.batch
import org.apache.Flink.api.Scala.ExecutionEnvironment
/**
* 单词计数之离线计算
* Created by xuwei.tech
*/
object BatchWordCountScala {
def main(args: Array[String]): Unit = {
val inputPath = "D:\\data\\file"
val outPut = "D:\\data\\result"
val env = ExecutionEnvironment.getExecutionEnvironment
val text = env.readTextFile(inputPath)
//引入隐式转换
import org.apache.Flink.api.Scala._
val counts = text.flatMap(_.toLowerCase.split("\\W+"))
.filter(_.nonEmpty)
.map((_,1))
.groupBy(0)
.sum(1)
counts.writeAsCsv(outPut,"\n"," ").setParallelism(1)
env.execute("batch word count")
}
}首先，代码中指定的inputPath是D:\\data\\file目录，我们需要在这个目录下面创建一些文件，并在文件中输入一些单词。
D:\data\file>dir
2018/03/20 09:01 24 a.txt
D:\data\file>type a.txt
hello a hello b
hello a然后，在IDEA中运行程序代码，产生的结果会被存储到outPut指定的D:\\data\\result文件中。
D:\data>type result
hello 3
b 1
a 2
