书名: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