高级大数据工程师带你一起学习Hdoop生态Flink基础原理保姆级教程
第1章Flink简介
1.1初识Flink
Flink起源于Stratosphere项目，Stratosphere是在2010~2014年由3所地处柏林的大学和欧洲的一些其他的大学共同进行的研究项目，2014年4月Stratosphere的代码被复制并捐赠给了Apache软件基金会，参加这个孵化项目的初始成员是Stratosphere系统的核心开发人员，2014年12月，Flink一跃成为Apache软件基金会的顶级项目。
在德语中，Flink一词表示快速和灵巧，项目采用一只松鼠的彩色图案作为logo，这不仅是因为松鼠具有快速和灵巧的特点，还因为柏林的松鼠有一种迷人的红棕色，而Flink的松鼠logo拥有可爱的尾巴，尾巴的颜色与Apache软件基金会的logo颜色相呼应，也就是说，这是一只Apache风格的松鼠。
Flink项目的理念是：“Apache Flink是为分布式、高性能、随时可用以及准确的流处理应用程序打造的开源流处理框架”。
Apache Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。Flink被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算。
1.2Flink的重要特点
1.2.1事件驱动型(Event-driven)
事件驱动型应用是一类具有状态的应用，它从一个或多个事件流提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。比较典型的就是以kafka为代表的消息队列几乎都是事件驱动型应用。(Flink的计算也是事件驱动型)
与之不同的就是SparkStreaming微批次，如图:
事件驱动型：
1.2.2流(flink)与批(spark)的世界观
批处理的特点是有界、大量，非常适合需要访问全套记录才能完成的计算工作，一般用于离线统计。
流处理的特点是无界、实时, 无需针对整个数据集执行操作，而是对通过系统传输的每个数据项执行操作，一般用于实时统计。
在spark的世界观中，一切都是由批次组成的，离线数据是一个大批次，而实时数据是由一个一个无限的小批次组成的。
而在flink的世界观中，一切都是由流组成的，离线数据是有界限的流，实时数据是一个没有界限的流，这就是所谓的有界流和无界流。
无界数据流：
无界数据流有一个开始但是没有结束，它们不会在生成时终止并提供数据，必须连续处理无界流，也就是说必须在获取后立即处理event。对于无界数据流我们无法等待所有数据都到达，因为输入是无界的，并且在任何时间点都不会完成。处理无界数据通常要求以特定顺序（例如事件发生的顺序）获取event，以便能够推断结果完整性。
有界数据流：
有界数据流有明确定义的开始和结束，可以在执行任何计算之前通过获取所有数据来处理有界流，处理有界流不需要有序获取，因为可以始终对有界数据集进行排序，有界流的处理也称为批处理。
1.2.3分层API
最底层级的抽象仅仅提供了有状态流，它将通过过程函数（Process Function）被嵌入到DataStream API中。底层过程函数（Process Function） 与 DataStream API 相集成，使其可以对某些特定的操作进行底层的抽象，它允许用户可以自由地处理来自一个或多个数据流的事件，并使用一致的容错的状态。除此之外，用户可以注册事件时间并处理时间回调，从而使程序可以处理复杂的计算。
实际上，大多数应用并不需要上述的底层抽象，而是针对核心API（Core APIs） 进行编程，比如DataStream API（有界或无界流数据）以及DataSet API（有界数据集）。这些API为数据处理提供了通用的构建模块，比如由用户定义的多种形式的转换（transformations），连接（joins），聚合（aggregations），窗口操作（windows）等等。DataSet API 为有界数据集提供了额外的支持，例如循环与迭代。这些API处理的数据类型以类（classes）的形式由各自的编程语言所表示。
Table API 是以表为中心的声明式编程，其中表可能会动态变化（在表达流数据时）。Table API遵循（扩展的）关系模型：表有二维数据结构（schema）（类似于关系数据库中的表），同时API提供可比较的操作，例如select、project、join、group-by、aggregate等。Table API程序声明式地定义了什么逻辑操作应该执行，而不是准确地确定这些操作代码的看上去如何。
尽管Table API可以通过多种类型的用户自定义函数（UDF）进行扩展，其仍不如核心API更具表达能力，但是使用起来却更加简洁（代码量更少）。除此之外，Table API程序在执行之前会经过内置优化器进行优化。
你可以在表与 DataStream/DataSet 之间无缝切换，以允许程序将 Table API 与 DataStream 以及 DataSet 混合使用。
Flink提供的最高层级的抽象是 SQL 。这一层抽象在语法与表达能力上与 Table API 类似，但是是以SQL查询表达式的形式表现程序。SQL抽象与Table API交互密切，同时SQL查询可以直接在Table API定义的表上执行。
目前Flink作为批处理还不是主流，不如Spark成熟，所以DataSet使用的并不是很多。Flink Table API和Flink SQL也并不完善，大多都由各大厂商自己定制。所以我们主要学习DataStream API的使用。实际上Flink作为最接近Google DataFlow模型的实现，是流批统一的观点，所以基本上使用DataStream就可以了。
2020年12月8日发布的1.12.0版本, 已经完成实现了真正的流批一体. 写好的一套代码, 即可以处理流式数据, 也可以处理离线数据. 这个与前面版本的处理有界流的方式是不一样的, Flink专门对批处理数据做了优化处理.
目前最新版本是2021年5月28日发布最新稳定版本为:1.13.1
1.3Spark or Flink
Spark 和 Flink 一开始都拥有着同一个梦想，他们都希望能够用同一个技术把流处理和批处理统一起来，但他们走了完全不一样的两条路前者是以批处理的技术为根本，并尝试在批处理之上支持流计算；后者则认为流计算技术是最基本的，在流计算的基础之上支持批处理。正因为这种架构上的不同，今后二者在能做的事情上会有一些细微的区别。比如在低延迟场景，Spark 基于微批处理的方式需要同步会有额外开销，因此无法在延迟上做到极致。在大数据处理的低延迟场景，Flink 已经有非常大的优势。
Spark和Flink的主要差别就在于计算模型不同。Spark采用了微批处理模型，而Flink采用了基于操作符的连续流模型。因此，对Apache Spark和Apache Flink的选择实际上变成了计算模型的选择，而这种选择需要在延迟、吞吐量和可靠性等多个方面进行权衡。
如果企业中非要技术选型从Spark和Flink这两个主流框架中选择一个来进行流数据处理，我们推荐使用Flink，主（显而）要（易见）的原因为：
Flink灵活的窗口
Exactly Once语义保证
事件时间(event-time)语义(处理乱序数据或者延迟数据)
这两个原因可以大大的解放程序员, 加快编程效率, 把本来需要程序员花大力气手动完成的工作交给框架完成，棒棒哒，点赞！！！
1.4Flink的应用
1.4.1应用Flink的场景
事件驱动型应用
事件驱动型应用是一类具有状态的应用，它从一个或多个事件流提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。
事件驱动型应用是在计算存储分离的传统应用基础上进化而来。在传统架构中，应用需要读写远程事务型数据库。
相反，事件驱动型应用是基于状态化流处理来完成。在该设计中，数据和计算不会分离，应用只需访问本地（内存或磁盘）即可获取数据。系统容错性的实现依赖于定期向远程持久化存储写入 checkpoint。下图描述了传统应用和事件驱动型应用架构的区别。
事件驱动型应用无须查询远程数据库，本地数据访问使得它具有更高的吞吐和更低的延迟。而由于定期向远程持久化存储的 checkpoint 工作可以异步、增量式完成，因此对于正常事件处理的影响甚微。事件驱动型应用的优势不仅限于本地数据访问。传统分层架构下，通常多个应用会共享同一个数据库，因而任何对数据库自身的更改（例如：由应用更新或服务扩容导致数据布局发生改变）都需要谨慎协调。反观事件驱动型应用，由于只需考虑自身数据，因此在更改数据表示或服务扩容时所需的协调工作将大大减少。
数据分析应用
Apache Flink 同时支持流式及批量分析应用。
Flink 为持续流式分析和批量分析都提供了良好的支持。具体而言，它内置了一个符合 ANSI 标准的 SQL 接口，将批、流查询的语义统一起来。无论是在记录事件的静态数据集上还是实时事件流上，相同 SQL 查询都会得到一致的结果。同时 Flink 还支持丰富的用户自定义函数，允许在 SQL 中执行定制化代码。如果还需进一步定制逻辑，可以利用 Flink DataStream API 和 DataSet API 进行更低层次的控制。此外，Flink 的 Gelly 库为基于批量数据集的大规模高性能图分析提供了算法和构建模块支持。
数据管道应用
提取-转换-加载（ETL）是一种在存储系统之间进行数据转换和迁移的常用方法。ETL 作业通常会周期性地触发，将数据从事务型数据库拷贝到分析型数据库或数据仓库。
数据管道和 ETL 作业的用途相似，都可以转换、丰富数据，并将其从某个存储系统移动到另一个。但数据管道是以持续流模式运行，而非周期性触发。因此它支持从一个不断生成数据的源头读取记录，并将它们以低延迟移动到终点。例如：数据管道可以用来监控文件系统目录中的新文件，并将其数据写入事件日志；另一个应用可能会将事件流物化到数据库或增量构建和优化查询索引。
下图描述了周期性 ETL 作业和持续数据管道的差异。
1.4.2应用Flink的行业
电商和市场营销
数据报表、广告投放
物联网（IOT）
传感器实时数据采集和显示、实时报警，交通运输业
物流配送和服务业
订单状态实时更新、通知信息推送、电信业基站流量调配
银行和金融业
实时结算和通知推送，实时检测异常行为
1.4.3应用Flink的企业
巨型大家经常用的电商，几乎涵盖大家基本生活的最大企业！
第2章Flink快速上手
2.1创建maven项目
POM文件中添加需要的依赖:
<properties>
<flink.version>1.13.1</flink.version>
<scala.binary.version>2.12</scala.binary.version>
<slf4j.version>1.7.30</slf4j.version>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>${slf4j.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>${slf4j.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-to-slf4j</artifactId>
<version>2.14.0</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.2.4</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<artifactSet>
<excludes>
<exclude>com.google.code.findbugs:jsr305</exclude>
<exclude>org.slf4j:*</exclude>
<exclude>log4j:*</exclude>
</excludes>
</artifactSet>
<filters>
<filter>
<!-- Do not copy the signatures in the META-INF folder.
Otherwise, this might cause SecurityExceptions when using the JAR. -->
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
<transformers combine.children="append">
<transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer">
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
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
配置idea, 运行的时候包括provided scope
2.2批处理WordCount
// 1. 创建执行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
// 2. 从文件读取数据 按行读取(存储的元素就是每行的文本)
DataSource<String> lineDS = env.readTextFile("input/words.txt");
// 3. 转换数据格式
FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDS
.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
String[] split = line.split(" ");
for (String word : split) {
out.collect(Tuple2.of(word, 1L));
}
})
.returns(Types.TUPLE(Types.STRING, Types.LONG)); // 当Lambda表达式使用 java 泛型的时候, 由于泛型擦除的存在, 需要显示的声明类型信息
// 4. 按照 word 进行分组
UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG = wordAndOne.groupBy(0);
// 5. 分组内聚合统计
AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1);
// 6. 打印结果
sum.print();
2.3流处理WordCount
2.3.1有界流
// 1. 创建流式执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 读取文件
DataStreamSource<String> lineDSS = env.readTextFile("input/words.txt");
// 3. 转换数据格式
SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = lineDSS
.flatMap((String line, Collector<String> words) -> {
Arrays.stream(line.split(" ")).forEach(words::collect);
})
.returns(Types.STRING)
.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));
// 4. 分组
KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne
.keyBy(t -> t.f0);
// 5. 求和
SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS
.sum(1);
// 6. 打印
result.print();
// 7. 执行
env.execute();
2.3.2无界流
// 1. 创建流式执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 读取文件
DataStreamSource<String> lineDSS = env.socketTextStream("hadoop162", 9999);
// 3. 转换数据格式
SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = lineDSS
.flatMap((String line, Collector<String> words) -> {
Arrays.stream(line.split(" ")).forEach(words::collect);
})
.returns(Types.STRING)
.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));
// 4. 分组
KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne
.keyBy(t -> t.f0);
// 5. 求和
SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS
.sum(1);
// 6. 打印
result.print();
// 7. 执行
env.execute();
第3章Flink部署
3.1开发模式
咱们前面在idea中运行Flink程序的方式就是开发模式.
3.2local-cluster模式
Flink中的Local-cluster(本地集群)模式,主要用于测试, 学习.
3.2.1local-cluster模式配置
local-cluster模式基本属于零配置.
1.上传Flink的安装包flink-1.13.1-bin-scala_2.12.tgz到hadoop162
2.解压
tar -zxvf flink-1.13.1-bin-scala_2.12.tgz -C /opt/module
3.进入目录/opt/module, 复制flink-local
cd /opt/module
cp -r flink-1.13.1 flink-local
3.2.2在local-cluster模式下运行无界的WordCount
1.打包idea中的应用
2.把不带依赖的jar包上传到目录/opt/module/flink-local下
3.启动本地集群
bin/start-cluster.sh
4.在hadoop162中启动netcat
nc -lk 9999
注意: 如果没有安装netcat需要先安装:
sudo yum install -y nc
5.命令行提交Flink应用
bin/flink run -m hadoop162:8081 -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
6.在浏览器中查看应用执行情况
http://hadoop162:8081
7.也可以在log日志查看执行结果
cat flink-atguigu-taskexecutor-0-hadoop162.out
8.也可以在WEB UI提交应用
3.3Standalone模式
Standalone模式又叫独立集群模式.
3.3.1Standalone模式配置
1.复制flink-standalone
cp -r flink-1.13.1 flink-standalone
2.修改配置文件:flink-conf.yaml
jobmanager.rpc.address: hadoop162
3.修改配置文件:workers
hadoop163
hadoop164
4.分发flink-standalone到其他节点
3.3.2Standalone模式运行无界流WorkCount
1.启动standalone集群
bin/start-cluster.sh
2.命令行提交Flink应用
bin/flink run -m hadoop162:8081 -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
3.查看执行情况与本地集群一致.
4.也支持Web UI界面提交Flink应用
3.3.3Standalone高可用(HA)
任何时候都有一个 主 JobManager 和多个备用 JobManagers，以便在主节点失败时有备用 JobManagers 来接管集群。这保证了没有单点故障，一旦备 JobManager 接管集群，作业就可以正常运行。主备 JobManager 实例之间没有明显的区别。每个 JobManager 都可以充当主备节点。
1.修改配置文件: link-conf.yaml
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop162:8020/flink/standalone/ha
high-availability.zookeeper.quorum: hadoop162:2181,hadoop163:2181,hadoop164:2181
high-availability.zookeeper.path.root: /flink-standalone
high-availability.cluster-id: /cluster_atguigu
2.修改配置文件: masters
hadoop162:8081
hadoop163:8081
3.分发修改的后配置文件到其他节点
4.在/etc/profile.d/my.sh中配置环境变量
export HADOOP_CLASSPATH=hadoop classpath
注意:
需要提前保证HAOOP_HOME环境变量配置成功
分发到其他节点
5.首先启动dfs集群和zookeeper集群
6.启动standalone HA集群
bin/start-cluster.sh
7.可以分别访问
http://hadoop162:8081
http://hadoop163:8081
8.在zkCli.sh中查看谁是leader
get /flink-standalone/cluster_atguigu/leader/rest_server_lock
杀死hadoop162上的Jobmanager, 再看leader
注意: 不管是不是leader从WEB UI上看不到区别, 并且都可以与之提交应用.
3.4Yarn模式
独立部署（Standalone）模式由Flink自身提供计算资源，无需其他框架提供资源，这种方式降低了和其他第三方资源框架的耦合性，独立性非常强。但是你也要记住，Flink主要是计算框架，而不是资源调度框架，所以本身提供的资源调度并不是它的强项，所以还是和其他专业的资源调度框架集成更靠谱，所以接下来我们来学习在强大的Yarn环境中Flink是如何使用的。（其实是因为在国内工作中，Yarn使用的非常多）
把Flink应用提交给Yarn的ResourceManager, Yarn的ResourceManager会申请容器从Yarn的NodeManager上面. Flink会创建JobManager和TaskManager在这些容器上.Flink会根据运行在JobManger上的job的需要的slot的数量动态的分配TaskManager资源
3.4.1Yarn模式配置
1.复制flink-yarn
cp -r flink-1.13.1 flink-yarn
2.配置环境变量HADOOP_CLASSPATH, 如果前面已经配置可以忽略.
在/etc/profile.d/my.sh中配置
export HADOOP_CLASSPATH=hadoop classpath
3.4.2Yarn运行无界流WordCount
1.启动hadoop集群(hdfs, yarn)
2.运行无界流
bin/flink run -t yarn-per-job -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
3.在yarn的ResourceManager界面查看执行情况
3.4.3Flink on Yarn的3种部署模式
Flink提供了yarn上运行的3模式，分别为Application Mode(1.12), Session-Cluster和Per-Job-Cluster模式。
Session-Cluster
Session-Cluster模式需要先启动Flink集群，向Yarn申请资源。以后提交任务都向这里提交。这个Flink集群会常驻在yarn集群中，除非手工停止。
在向Flink集群提交Job的时候, 如果资源被用完了,则新的Job不能正常提交.
缺点: 如果提交的作业中有长时间执行的大作业, 占用了该Flink集群的所有资源, 则后续无法提交新的job.
所以, Session-Cluster适合那些需要频繁提交的多个小Job, 并且执行时间都不长的Job.
Per-Job-Cluster
一个Job会对应一个Flink集群，每提交一个作业会根据自身的情况，都会单独向yarn申请资源，直到作业执行完成，一个作业的失败与否并不会影响下一个作业的正常提交和运行。独享Dispatcher和ResourceManager，按需接受资源申请；适合规模大长时间运行的作业。
每次提交都会创建一个新的flink集群，任务之间互相独立，互不影响，方便管理。任务执行完成之后创建的集群也会消失。
Application Mode
Application Mode会在Yarn上启动集群, 应用jar包的main函数(用户类的main函数)将会在JobManager上执行. 只要应用程序执行结束, Flink集群会马上被关闭. 也可以手动停止集群.
与Per-Job-Cluster的区别: 就是Application Mode下, 用户的main函数是在集群中(job manager)执行的
官方建议:
出于生产的需求, 我们建议使用Per-job or Application Mode,因为他们给应用提供了更好的隔离!
3.4.4Per-Job-Cluster模式执行无界流WordCount
bin/flink run -d -t yarn-per-job -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
3.4.5Session-Cluster模式执行无界流WordCount
1.启动一个Flink-Session
bin/yarn-session.sh -d
2.在Session上运行Job
bin/flink run -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
会自动找到你的yarn-session启动的Flink集群.也可以手动指定你的yarn-session集群:
bin/flink run -t yarn-session -Dyarn.application.id=application_XXXX_YY ./flink-prepare-1.0-SNAPSHOT.ja
注意: application_XXXX_YY 指的是在yarn上启动的yarn应用
33.4.6Application Mode模式执行无界流WordCount
bin/flink run-application -t yarn-application -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
3.4.7Yarn模式高可用
Yarn模式的高可用和Standalone模式的高可用原理不一样.
Standalone模式中, 同时启动多个Jobmanager, 一个为leader其他为standby, 当leader挂了, 其他的才会有一个成为leader.
yarn的高可用是同时只启动一个Jobmanager, 当这个Jobmanager挂了之后, yarn会再次启动一个, 其实是利用的yarn的重试次数来实现的高可用.
1.在yarn-site.xml中配置
<property>
<name>yarn.resourcemanager.am.max-attempts</name>
<value>4</value>
<description>
The maximum number of application master execution attempts.
</description>
</property>
注意: 配置完不要忘记分发, 和重启yarn
2.在flink-conf.yaml中配置
yarn.application-attempts: 3
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop162:8020/flink/yarn/ha
high-availability.zookeeper.quorum: hadoop162:2181,hadoop163:2181,hadoop164:2181
high-availability.zookeeper.path.root: /flink-yarn
3.启动yarn-session
4.杀死Jobmanager, 查看他的复活情况
注意: yarn-site.xml中是它活的次数的上限, flink-conf.xml中的次数应该小于这个值.
3.5 Scala REPL
如果flink的scala版本是2.12则不提供这个repl
scala 交互环境.
1.local模式启动 REPL
/opt/module/flink-local » bin/start-scala-shell.sh local
2.yarn-session 模式启动
先启动一个yarn-session, 然后就可以把shell跑在yarn-session上了
bin/start-scala-shell.sh yarn
3.6K8S & Mesos模式
Mesos是Apache下的开源分布式资源管理框架，它被称为是分布式系统的内核,在Twitter得到广泛使用,管理着Twitter超过30,0000台服务器上的应用部署，但是在国内，依然使用着传统的Hadoop大数据框架，所以国内使用mesos框架的并不多，这里我们就不做过多讲解了。
容器化部署时目前业界很流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes（k8s），而Flink也在最近的版本中支持了k8s部署模式。这里我们也不做过多的讲解.
第4章Flink运行架构
4.1运行架构
https://ci.apache.org/projects/flink/flink-docs-release-1.13/fig/processes.svg
Flink运行时包含2种进程:1个JobManager和至少1个TaskManager
4.1.1客户端
严格上说, 客户端不是运行和程序执行的一部分, 而是用于准备和发送dataflow到JobManager. 然后客户端可以断开与JobManager的连接(detached mode), 也可以继续保持与JobManager的连接(attached mode)
客户端作为触发执行的java或者scala代码的一部分运行, 也可以在命令行运行:bin/flink run …
4.1.2JobManager
控制一个应用程序执行的主进程，也就是说，每个应用程序都会被一个的JobManager所控制执行。
JobManager会先接收到要执行的应用程序，这个应用程序会包括：作业图（JobGraph）、逻辑数据流图（logical dataflow graph）和打包了所有的类、库和其它资源的JAR包。
JobManager会把JobGraph转换成一个物理层面的数据流图，这个图被叫做“执行图”（ExecutionGraph），包含了所有可以并发执行的任务。JobManager会向资源管理器（ResourceManager）请求执行任务必要的资源，也就是任务管理器（TaskManager）上的插槽（slot）。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的TaskManager上。
而在运行过程中，JobManager会负责所有需要中央协调的操作，比如说检查点（checkpoints）的协调。
这个进程包含3个不同的组件
4.1.2.1ResourceManager
负责资源的管理，在整个 Flink 集群中只有一个 ResourceManager. 注意这个ResourceManager不是Yarn中的ResourceManager, 是Flink中内置的, 只是赶巧重名了而已.
主要负责管理任务管理器（TaskManager）的插槽（slot），TaskManger插槽是Flink中定义的处理资源单元。
当JobManager申请插槽资源时，ResourceManager会将有空闲插槽的TaskManager分配给JobManager。如果ResourceManager没有足够的插槽来满足JobManager的请求，它还可以向资源提供平台发起会话，以提供启动TaskManager进程的容器。另外，ResourceManager还负责终止空闲的TaskManager，释放计算资源。
4.1.2.2Dispatcher
负责接收用户提供的作业，并且负责为这个新提交的作业启动一个新的JobMaster 组件. Dispatcher也会启动一个Web UI，用来方便地展示和监控作业执行的信息。Dispatcher在架构中可能并不是必需的，这取决于应用提交运行的方式。
4.1.2.3JobMaster
JobMaster负责管理单个JobGraph的执行.多个Job可以同时运行在一个Flink集群中, 每个Job都有一个自己的JobMaster.
4.1.3TaskManager
Flink中的工作进程。通常在Flink中会有多个TaskManager运行，每一个TaskManager都包含了一定数量的插槽（slots）。插槽的数量限制了TaskManager能够执行的任务数量。
启动之后，TaskManager会向资源管理器注册它的插槽；收到资源管理器的指令后，TaskManager就会将一个或者多个插槽提供给JobManager调用。JobManager就可以向插槽分配任务（tasks）来执行了。
在执行过程中，一个TaskManager可以跟其它运行同一应用程序的TaskManager交换数据。
4.2核心概念
4.2.1TaskManager与Slots
Flink中每一个worker(TaskManager)都是一个JVM进程，它可能会在独立的线程上执行一个Task。为了控制一个worker能接收多少个task，worker通过Task Slot来进行控制（一个worker至少有一个Task Slot）。
这里的Slot如何来理解呢？很多的文章中经常会和Spark框架进行类比，将Slot类比为Core，其实简单这么类比是可以的，可实际上，可以考虑下，当Spark申请资源后，这个Core执行任务时有可能是空闲的，但是这个时候Spark并不能将这个空闲下来的Core共享给其他Job使用，所以这里的Core是Job内部共享使用的。接下来我们再回想一下，之前在Yarn Session-Cluster模式时，其实是可以并行执行多个Job的，那如果申请两个Slot，而执行Job时，只用到了一个，剩下的一个怎么办？那我们自认而然就会想到可以将这个Slot给并行的其他Job，对吗？所以Flink中的Slot和Spark中的Core还是有很大区别的。
每个task slot表示TaskManager拥有资源的一个固定大小的子集。假如一个TaskManager有三个slot，那么它会将其管理的内存分成三份给各个slot。资源slot化意味着一个task将不需要跟来自其他job的task竞争被管理的内存，取而代之的是它将拥有一定数量的内存储备。需要注意的是，这里不会涉及到CPU的隔离，slot目前仅仅用来隔离task的受管理的内存。
4.2.2Parallelism（并行度）
一个特定算子的子任务（subtask）的个数被称之为这个算子的并行度（parallelism），一般情况下，一个流程序的并行度，可以认为就是其所有算子中最大的并行度。一个程序中，不同的算子可能具有不同的并行度。
Stream在算子之间传输数据的形式可以是one-to-one(forwarding)的模式也可以是redistributing的模式，具体是哪一种形式，取决于算子的种类。
One-to-one：
stream(比如在source和map operator之间)维护着分区以及元素的顺序。那意味着flatmap 算子的子任务看到的元素的个数以及顺序跟source 算子的子任务生产的元素的个数、顺序相同，map、fliter、flatMap等算子都是one-to-one的对应关系。类似于spark中的窄依赖
Redistributing：
stream(map()跟keyBy/window之间或者keyBy/window跟sink之间)的分区会发生改变。每一个算子的子任务依据所选择的transformation发送数据到不同的目标任务。例如，keyBy()基于hashCode重分区、broadcast和rebalance会随机重新分区，这些算子都会引起redistribute过程，而redistribute过程就类似于Spark中的shuffle过程。类似于spark中的宽依赖
4.2.3Task与SubTask
一个算子就是一个Task. 一个算子的并行度是几, 这个Task就有几个SubTask
4.2.4Operator Chains（任务链）
相同并行度的one to one操作，Flink将这样相连的算子链接在一起形成一个task，原来的算子成为里面的一部分。 每个subtask被一个线程执行.
将算子链接成task是非常有效的优化：它能减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。链接的行为可以在编程API中进行指定。
4.2.5ExecutionGraph（执行图）
由Flink程序直接映射成的数据流图是StreamGraph，也被称为逻辑流图，因为它们表示的是计算逻辑的高级视图。为了执行一个流处理程序，Flink需要将逻辑流图转换为物理数据流图（也叫执行图），详细说明程序的执行方式。
Flink 中的执行图可以分成四层：StreamGraph -> JobGraph -> ExecutionGraph -> Physical Graph。
StreamGraph：
是根据用户通过 Stream API 编写的代码生成的最初的图。用来表示程序的拓扑结构。
JobGraph：
StreamGraph经过优化后生成了 JobGraph，是提交给 JobManager 的数据结构。主要的优化为: 将多个符合条件的节点 chain 在一起作为一个节点，这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。
ExecutionGraph：
JobManager 根据 JobGraph 生成ExecutionGraph。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。
Physical Graph：
JobManager 根据 ExecutionGraph 对 Job 进行调度后，在各个TaskManager 上部署 Task 后形成的“图”，并不是一个具体的数据结构。
2个并发度（Source为1个并发度）的 SocketTextStreamWordCount 四层执行图的演变过程
env.socketTextStream().flatMap(…).keyBy(0).sum(1).print();
4.3提交流程
4.3.1高级视角提交流程(通用提交流程)
我们来看看当一个应用提交执行时，Flink的各个组件是如何交互协作的：
4.3.2yarn-cluster提交流程per-job
1.Flink任务提交后，Client向HDFS上传Flink的Jar包和配置
2.向Yarn ResourceManager提交任务，ResourceManager分配Container资源
3.通知对应的NodeManager启动ApplicationMaster，ApplicationMaster启动后加载Flink的Jar包和配置构建环境，然后启动JobManager
4.ApplicationMaster向ResourceManager申请资源启动TaskManager
5.ResourceManager分配Container资源后，由ApplicationMaster通知资源所在节点的NodeManager启动TaskManager
6.NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager
7.TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务。
第5章Flink流处理核心编程
和其他所有的计算框架一样，Llink也有一些基础的开发步骤以及基础，核心的API，从开发步骤的角度来讲，主要分为四大部分
5.1Environment
Flink Job在提交执行计算时，需要首先建立和Flink框架之间的联系，也就指的是当前的flink运行环境，只有获取了环境信息，才能将task调度到不同的taskManager执行。而这个环境对象的获取方式相对比较简单
// 批处理环境
ExecutionEnvironment benv = ExecutionEnvironment.getExecutionEnvironment();
// 流式数据处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
5.2Source
Flink框架可以从不同的来源获取数据，将数据提交给框架进行处理, 我们将获取数据的来源称之为数据源(Source)。
5.2.1准备工作
1.导入注解工具依赖, 方便生产POJO类
<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.16</version>
<scope>provided</scope>
</dependency>
2.准备一个WaterSensor类方便演示
package com.atguigu.flink.java.chapter_5;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
*
*
* 水位传感器：用于接收水位数据
*
* id:传感器编号
* ts:时间戳
* vc:水位
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class WaterSensor {
private String id;
private Long ts;
private Integer vc;
}
5.2.2从Java的集合中读取数据
一般情况下，可以将数据临时存储到内存中，形成特殊的数据结构后，作为数据源使用。这里的数据结构采用集合类型是比较普遍的。
package com.atguigu.flink.java.chapter_5;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.Arrays;
import java.util.List;
/**
*
*
*/
public class Flink01_Source_Collection {
public static void main(String[] args) throws Exception {
List<WaterSensor> waterSensors = Arrays.asList(
new WaterSensor("ws_001", 1577844001L, 45),
new WaterSensor("ws_002", 1577844015L, 43),
new WaterSensor("ws_003", 1577844020L, 42));
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.fromCollection(waterSensors)
.print();
env.execute();
}
}
5.2.3从文件读取数据
package com.atguigu.flink.java.chapter_5;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
*
*
*/
public class Flink02_Source_File {
public static void main(String[] args) throws Exception {
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.readTextFile("input")
.print();
env.execute();
}
}
说明:
1.参数可以是目录也可以是文件
2.路径可以是相对路径也可以是绝对路径
3.相对路径是从系统属性user.dir获取路径:
idea下是project的根目录,
standalone模式下指的当前的用户的home目录
yarn: ?
4.也可以从hdfs目录下读取, 使用路径:hdfs://…, 由于Flink没有提供hadoop相关依赖, 需要pom中添加相关依赖:
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>3.1.3</version>
<scope>provided</scope>
</dependency>
5.2.4从Socket读取数据
参考第1章无界流读取
5.2.5从Kafka读取数据
添加相应的依赖
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.12</artifactId>
<version>1.13.1</version>
</dependency>
参考代码
package com.atguigu.flink.java.chapter_5;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import java.util.Properties;
/**
*
*
*/
public class Flink03_Source_Kafka {
public static void main(String[] args) throws Exception {
// 0.Kafka相关配置
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "hadoop102:9092,hadoop103:9092,hadoop104:9092");
properties.setProperty("group.id", "Flink01_Source_Kafka");
properties.setProperty("auto.offset.reset", "latest");
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.addSource(new FlinkKafkaConsumer<>("sensor", new SimpleStringSchema(), properties))
.print("kafka source");
env.execute();
}
}
5.2.6自定义Source
大多数情况下，前面的数据源已经能够满足需要，但是难免会存在特殊情况的场合，所以flink也提供了能自定义数据源的方式.
/**
package com.atguigu.flink.java.chapter_5;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
*
*
*/
public class Flink04_Source_Custom {
public static void main(String[] args) throws Exception {
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.addSource(new MySource("hadoop102", 9999))
.print();
env.execute();
}
public static class MySource implements SourceFunction<WaterSensor> {
private String host;
private int port;
private volatile boolean isRunning = true;
private Socket socket;
public MySource(String host, int port) {
this.host = host;
this.port = port;
}
@Override
public void run(SourceContext<WaterSensor> ctx) throws Exception {
// 实现一个从socket读取数据的source
socket = new Socket(host, port);
BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
String line = null;
while (isRunning && (line = reader.readLine()) != null) {
String[] split = line.split(",");
ctx.collect(new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2])));
}
}
/**
* 大多数的source在run方法内部都会有一个while循环,
* 当调用这个方法的时候, 应该可以让run方法中的while循环结束
*/
@Override
public void cancel() {
isRunning = false;
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
/*
sensor_1,1607527992000,20
sensor_1,1607527993000,40
sensor_1,1607527994000,50
*/
5.3Transform
转换算子可以把一个或多个DataStream转成一个新的DataStream.程序可以把多个复杂的转换组合成复杂的数据流拓扑.
5.3.1map
作用
将数据流中的数据进行转换, 形成新的数据流，消费一个元素并产出一个元素
参数
lambda表达式或MapFunction实现类
返回
DataStream → DataStream
示例
得到一个新的数据流: 新的流的元素是原来流的元素的平方
匿名内部类对象
package com.atguigu.flink.java.chapter_5.transform;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
*
*
*/
public class Flink01_TransForm_Map_Anonymous {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.fromElements(1, 2, 3, 4, 5)
.map(new MapFunction<Integer, Integer>() {
@Override
public Integer map(Integer value) throws Exception {
return value * value;
}
})
.print();
env.execute();
}
}
Lambda表达式表达式
env
.fromElements(1, 2, 3, 4, 5)
.map(ele -> ele * ele)
.print();
静态内部类
package com.atguigu.flink.java.chapter_5.transform;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
*
*
*/
public class Flink01_TransForm_Map_StaticClass {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.fromElements(1, 2, 3, 4, 5)
.map(new MyMapFunction())
.print();
env.execute();
}
public static class MyMapFunction implements MapFunction<Integer, Integer> {
@Override
public Integer map(Integer value) throws Exception {
return value * value;
}
}
}
Rich…Function类
所有Flink函数类都有其Rich版本。它与常规函数的不同在于，可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。也有意味着提供了更多的，更丰富的功能。例如：RichMapFunction
package com.atguigu.flink.java.chapter_5.transform;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
*
*
*/
public class Flink01_TransForm_Map_RichMapFunction {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(5);
env
.fromElements(1, 2, 3, 4, 5)
.map(new MyRichMapFunction()).setParallelism(2)
.print();
env.execute();
}
public static class MyRichMapFunction extends RichMapFunction<Integer, Integer> {
// 默认生命周期方法, 初始化方法, 在每个并行度上只会被调用一次
@Override
public void open(Configuration parameters) throws Exception {
System.out.println("open ... 执行一次");
}
// 默认生命周期方法, 最后一个方法, 做一些清理工作, 在每个并行度上只调用一次
@Override
public void close() throws Exception {
System.out.println("close ... 执行一次");
}
@Override
public Integer map(Integer value) throws Exception {
System.out.println("map ... 一个元素执行一次");
return value * value;
}
}
}
1.默认生命周期方法, 初始化方法, 在每个并行度上只会被调用一次, 而且先被调用
2.默认生命周期方法, 最后一个方法, 做一些清理工作, 在每个并行度上只调用一次, 而且是最后被调用
3.getRuntimeContext()方法提供了函数的RuntimeContext的一些信息，例如函数执行的并行度，任务的名字，以及state状态. 开发人员在需要的时候自行调用获取运行时上下文对象.
5.3.2flatMap
作用
消费一个元素并产生零个或多个元素
参数
FlatMapFunction实现类
返回
DataStream → DataStream
示例
匿名内部类写法
// 新的流存储每个元素的平方和3次方
env
.fromElements(1, 2, 3, 4, 5)
.flatMap(new FlatMapFunction<Integer, Integer>() {
@Override
public void flatMap(Integer value, Collector<Integer> out) throws Exception {
out.collect(value * value);
out.collect(value * value * value);
}
})
.print();
Lambda表达式写法
env
.fromElements(1, 2, 3, 4, 5)
.flatMap((Integer value, Collector<Integer> out) -> {
out.collect(value * value);
out.collect(value * value * value);
}).returns(Types.INT)
.print();
说明: 在使用Lambda表达式表达式的时候, 由于泛型擦除的存在, 在运行的时候无法获取泛型的具体类型, 全部当做Object来处理, 及其低效, 所以Flink要求当参数中有泛型的时候, 必须明确指定泛型的类型.
5.3.3filter
作用
根据指定的规则将满足条件（true）的数据保留，不满足条件(false)的数据丢弃
参数
FlatMapFunction实现类
返回
DataStream → DataStream
示例
匿名内部类写法
// 保留偶数, 舍弃奇数
env
.fromElements(10, 3, 5, 9, 20, 8)
.filter(new FilterFunction<Integer>() {
@Override
public boolean filter(Integer value) throws Exception {
return value % 2 == 0;
}
})
.print();
Lambda表达式写法
env
.fromElements(10, 3, 5, 9, 20, 8)
.filter(value -> value % 2 == 0)
.print();
5.3.4keyBy
作用
把流中的数据分到不同的分区(并行度)中.具有相同key的元素会分到同一个分区中.一个分区中可以有多重不同的key.
在内部是使用的是key的hash分区来实现的.
参数
Key选择器函数: interface KeySelector<IN, KEY>
注意: 什么值不可以作为KeySelector的Key:
没有覆写hashCode方法的POJO, 而是依赖Object的hashCode. 因为这样分组没有任何的意义: 每个元素都会得到一个独立无二的组. 实际情况是:可以运行, 但是分的组没有意义.
任何类型的数组
返回
DataStream → KeyedStream
示例
匿名内部类写法
// 奇数分一组, 偶数分一组
env
.fromElements(10, 3, 5, 9, 20, 8)
.keyBy(new KeySelector<Integer, String>() {
@Override
public String getKey(Integer value) throws Exception {
return value % 2 == 0 ? "偶数" : "奇数";
}
})
.print();
env.execute();
Lambda表达式写法
env
.fromElements(10, 3, 5, 9, 20, 8)
.keyBy(value -> value % 2 == 0 ? "偶数" : "奇数")
.print();
5.3.5shuffle
作用
把流中的元素随机打乱. 对同一个组数据, 每次执行得到的结果都不同.
参数
无
返回
DataStream → DataStream
示例
env
.fromElements(10, 3, 5, 9, 20, 8)
.shuffle()
.print();
env.execute();
5.3.7connect
作用
在某些情况下，我们需要将两个不同来源的数据流进行连接，实现数据匹配，比如订单支付和第三方交易信息，这两个信息的数据就来自于不同数据源，连接后，将订单支付和第三方交易信息进行对账，此时，才能算真正的支付完成。
Flink中的connect算子可以连接两个保持他们类型的数据流，两个数据流被connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。
参数
另外一个流
返回
DataStream[A], DataStream[B] -> ConnectedStreams[A,B]
示例
DataStreamSource<Integer> intStream = env.fromElements(1, 2, 3, 4, 5);
DataStreamSource<String> stringStream = env.fromElements("a", "b", "c");
// 把两个流连接在一起: 貌合神离
ConnectedStreams<Integer, String> cs = intStream.connect(stringStream);
cs.getFirstInput().print("first");
cs.getSecondInput().print("second");
env.execute();
注意:
1.两个流中存储的数据类型可以不同
2.只是机械的合并在一起, 内部仍然是分离的2个流
3.只能2个流进行connect, 不能有第3个参与
5.3.8union
作用
对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream
示例
DataStreamSource<Integer> stream1 = env.fromElements(1, 2, 3, 4, 5);
DataStreamSource<Integer> stream2 = env.fromElements(10, 20, 30, 40, 50);
DataStreamSource<Integer> stream3 = env.fromElements(100, 200, 300, 400, 500);
// 把多个流union在一起成为一个流, 这些流中存储的数据类型必须一样: 水乳交融
stream1
.union(stream2)
.union(stream3)
.print();
connect与 union 区别：
1.union之前两个或多个流的类型必须是一样，connect可以不一样
2.connect只能操作两个流，union可以操作多个。
5.3.9简单滚动聚合算子
常见的滚动聚合算子
sum, min,max
minBy,maxBy
作用
KeyedStream的每一个支流做聚合。执行完成后，会将聚合的结果合成一个流返回，所以结果都是DataStream
参数
如果流中存储的是POJO或者scala的样例类, 参数使用字段名. 如果流中存储的是元组, 参数就是位置(基于0…)
返回
KeyedStream -> SingleOutputStreamOperator
示例
示例1
DataStreamSource<Integer> stream = env.fromElements(1, 2, 3, 4, 5);
KeyedStream<Integer, String> kbStream = stream.keyBy(ele -> ele % 2 == 0 ? "奇数" : "偶数");
kbStream.sum(0).print("sum");
kbStream.max(0).print("max");
kbStream.min(0).print("min");
示例2
ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 30));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
KeyedStream<WaterSensor, String> kbStream = env
.fromCollection(waterSensors)
.keyBy(WaterSensor::getId);
kbStream
.sum("vc")
.print("maxBy...");
注意:
分组聚合后, 理论上只能取分组字段和聚合结果, 但是Flink允许其他的字段也可以取出来, 其他字段默认情况是取的是这个组内第一个元素的字段值
示例3:
ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
KeyedStream<WaterSensor, String> kbStream = env
.fromCollection(waterSensors)
.keyBy(WaterSensor::getId);
kbStream
.maxBy("vc", false)
.print("max...");
env.execute();
注意:
maxBy和minBy可以指定当出现相同值的时候,其他字段是否取第一个. true表示取第一个, false表示取与最大值(最小值)同一行的.
5.3.10reduce
作用
一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。
为什么还要把中间值也保存下来? 考虑流式数据的特点: 没有终点, 也就没有最终的概念了. 任何一个中间的聚合结果都是值!
参数
interface ReduceFunction
返回
KeyedStream -> SingleOutputStreamOperator
示例
匿名内部类写法
ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
KeyedStream<WaterSensor, String> kbStream = env
.fromCollection(waterSensors)
.keyBy(WaterSensor::getId);
kbStream
.reduce(new ReduceFunction<WaterSensor>() {
@Override
public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
System.out.println("reducer function ...");
return new WaterSensor(value1.getId(), value1.getTs(), value1.getVc() + value2.getVc());
}
})
.print("reduce...");
env.execute();
Lambda表达式写法
kbStream
.reduce((value1, value2) -> {
System.out.println("reducer function ...");
return new WaterSensor(value1.getId(), value1.getTs(), value1.getVc() + value2.getVc());
})
.print("reduce...");
注意:
1.聚合后结果的类型, 必须和原来流中元素的类型保持一致!
5.3.11process
作用
process算子在Flink算是一个比较底层的算子, 很多类型的流上都可以调用, 可以从流中获取更多的信息(不仅仅数据本身)
示例1: 在keyBy之前的流上使用
env
.fromCollection(waterSensors)
.process(new ProcessFunction<WaterSensor, Tuple2<String, Integer>>() {
@Override
public void processElement(WaterSensor value,
Context ctx,
Collector<Tuple2<String, Integer>> out) throws Exception {
out.collect(new Tuple2<>(value.getId(), value.getVc()));
}
})
.print();
示例2: 在keyBy之后的流上使用
env
.fromCollection(waterSensors)
.keyBy(WaterSensor::getId)
.process(new KeyedProcessFunction<String, WaterSensor, Tuple2<String, Integer>>() {
@Override
public void processElement(WaterSensor value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
out.collect(new Tuple2<>("key是:" + ctx.getCurrentKey(), value.getVc()));
}
})
.print();
5.3.12对流重新分区的几个算子
KeyBy
先按照key分组, 按照key的双重hash来选择后面的分区
shuffle
对流中的元素随机分区
rebalance
对流中的元素平均分布到每个区.当处理倾斜数据的时候, 进行性能优化
rescale
同 rebalance一样, 也是平均循环的分布数据. 但是要比rebalance更高效, 因为rescale不需要通过网络, 完全走的"管道"
5.4Sink
Sink有下沉的意思，在Flink中所谓的Sink其实可以表示为将数据存储起来的意思，也可以将范围扩大，表示将处理完的数据发送到指定的存储系统的输出操作.
之前我们一直在使用的print方法其实就是一种Sink
public DataStreamSink print(String sinkIdentifier) {
PrintSinkFunction printFunction = new PrintSinkFunction<>(sinkIdentifier, false);
return addSink(printFunction).name(“Print to Std. Out”);
}
Flink内置了一些Sink, 除此之外的Sink需要用户自定义!
5.4.1KafkaSink
添加Kafka Connector依赖
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.12</artifactId>
<version>1.13.1</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.75</version>
</dependency>
启动Kafka集群
Sink到Kafka的示例代码
package com.atguigu.flink.java.chapter_5.sink;
import com.alibaba.fastjson.JSON;
import com.atguigu.flink.java.chapter_5.WaterSensor;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.ArrayList;
/**
*
*
*/
public class Flink01_Sink_Kafka {
public static void main(String[] args) throws Exception {
ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
env
.fromCollection(waterSensors)
.map(JSON::toJSONString)
.addSink(new FlinkKafkaProducer<String>("hadoop102:9092", "topic_sensor", new SimpleStringSchema()));
env.execute();
}
}
在linux启动一个消费者, 查看是否收到数据
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_sensor
5.4.2RedisSink
添加Redis Connector依赖
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-redis_2.11</artifactId>
<version>1.1.5</version>
</dependency>
启动Redis服务器
Sink到Redis的示例代码
package com.atguigu.flink.java.chapter_5.sink;
import com.alibaba.fastjson.JSON;
import com.atguigu.flink.java.chapter_5.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import java.util.ArrayList;
/**
*
*
*/
public class Flink02_Sink_Redis {
public static void main(String[] args) throws Exception {
ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
// 连接到Redis的配置
FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
.setHost("hadoop102")
.setPort(6379)
.setMaxTotal(100)
.setTimeout(1000 * 10)
.build();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
env
.fromCollection(waterSensors)
.addSink(new RedisSink<>(redisConfig, new RedisMapper<WaterSensor>() {
/*
key value(hash)
"sensor" field value
sensor_1 {"id":"sensor_1","ts":1607527992000,"vc":20}
... ...
*/
@Override
public RedisCommandDescription getCommandDescription() {
// 返回存在Redis中的数据类型 存储的是Hash, 第二个参数是外面的key
return new RedisCommandDescription(RedisCommand.HSET, "sensor");
}
@Override
public String getKeyFromData(WaterSensor data) {
// 从数据中获取Key: Hash的Key
return data.getId();
}
@Override
public String getValueFromData(WaterSensor data) {
// 从数据中获取Value: Hash的value
return JSON.toJSONString(data);
}
}));
env.execute();
}
}
Redis查看是否收到数据
redis-cli --raw
注意:
发送了5条数据, redis中只有2条数据. 原因是hash的field的重复了, 后面的会把前面的覆盖掉
5.4.3ElasticsearchSink
添加Elasticsearch Connector依赖
org.apache.flink flink-connector-elasticsearch6_2.12 1.13.1 启动Elasticsearch集群 Sink到Elasticsearch的示例代码package com.atguigu.flink.java.chapter_5.sink;
import com.alibaba.fastjson.JSON;
import com.atguigu.flink.java.chapter_5.WaterSensor;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
/**
*
*
*/
public class Flink03_Sink_ES {
public static void main(String[] args) throws Exception {
ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
List<HttpHost> esHosts = Arrays.asList(
new HttpHost("hadoop102", 9200),
new HttpHost("hadoop103", 9200),
new HttpHost("hadoop104", 9200));
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
env
.fromCollection(waterSensors)
.addSink(new ElasticsearchSink.Builder<WaterSensor>(esHosts, new ElasticsearchSinkFunction<WaterSensor>() {
@Override
public void process(WaterSensor element, RuntimeContext ctx, RequestIndexer indexer) {
// 1. 创建es写入请求
IndexRequest request = Requests
.indexRequest("sensor")
.type("_doc")
.id(element.getId())
.source(JSON.toJSONString(element), XContentType.JSON);
// 2. 写入到es
indexer.add(request);
}
}).build());
env.execute();
}
}
Elasticsearch查看是否收到数据
注意
如果出现如下错误:
添加log4j2的依赖:
org.apache.logging.log4j
log4j-to-slf4j
2.14.0
如果是无界流, 需要配置bulk的缓存
esSinkBuilder.setBulkFlushMaxActions(1);
5.4.4自定义Sink
如果Flink没有提供给我们可以直接使用的连接器，那我们如果想将数据存储到我们自己的存储设备中，怎么办？
我们自定义一个到Mysql的Sink
在mysql中创建数据库和表
create database test;
use test;
CREATE TABLE `sensor` (
`id` varchar(20) NOT NULL,
`ts` bigint(20) NOT NULL,
`vc` int(11) NOT NULL,
PRIMARY KEY (`id`,`ts`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
导入Mysql驱动
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.49</version>
</dependency>
写到Mysql的自定义Sink示例代码
package com.atguigu.flink.java.chapter_5.sink;
import com.atguigu.flink.java.chapter_5.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList;
/**
*
*
*/
public class Flink04_Sink_Custom {
public static void main(String[] args) throws Exception {
ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
env.fromCollection(waterSensors)
.addSink(new RichSinkFunction<WaterSensor>() {
private PreparedStatement ps;
private Connection conn;
@Override
public void open(Configuration parameters) throws Exception {
conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test?useSSL=false", "root", "aaaaaa");
ps = conn.prepareStatement("insert into sensor values(?, ?, ?)");
}
@Override
public void close() throws Exception {
ps.close();
conn.close();
}
@Override
public void invoke(WaterSensor value, Context context) throws Exception {
ps.setString(1, value.getId());
ps.setLong(2, value.getTs());
ps.setInt(3, value.getVc());
ps.execute();
}
});
env.execute();
}
}
5.5执行模式(Execution Mode)
Flink从1.12.0上对流式API新增一项特性:可以根据你的使用情况和Job的特点, 可以选择不同的运行时执行模式(runtime execution modes).
流式API的传统执行模式我们称之为STREAMING 执行模式, 这种模式一般用于无界流, 需要持续的在线处理
1.12.0新增了一个BATCH执行模式, 这种执行模式在执行方式上类似于MapReduce框架. 这种执行模式一般用于有界数据.
默认是使用的STREAMING 执行模式
5.5.1选择执行模式
BATCH执行模式仅仅用于有界数据, 而STREAMING 执行模式可以用在有界数据和无界数据.
一个公用的规则就是: 当你处理的数据是有界的就应该使用BATCH执行模式, 因为它更加高效. 当你的数据是无界的, 则必须使用STREAMING 执行模式, 因为只有这种模式才能处理持续的数据流.
5.5.2配置BATH执行模式
执行模式有3个选择可配:
1STREAMING(默认)
2BATCH
3AUTOMATIC
自动;如果数据离线数据, 则自动选择批处理, 如果是流数据, 则自动选流处理
通过命令行配置
bin/flink run -Dexecution.runtime-mode=BATCH …
通过代码配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
建议: 不要在运行时配置(代码中配置), 而是使用命令行配置, 因为这样会更灵活: 同一个应用即可以用于无界数据也可以用于有界数据
5.5.3有界数据用STREAMING和BATCH的区别
STREAMING模式下, 数据是来一条输出一次结果.
BATCH模式下, 数据处理完之后, 一次性输出结果.
下面展示WordCount的程序读取文件内容在不同执行模式下的执行结果对比:
流式模式
// 默认流式模式, 可以不用配置
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
批处理模式
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
自动模式
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
注意:
1.12.0版批模式下有个bug, 如果单词只有1个, 则不会输出.
1.13开始, 这个bug已经解决了