flink_01

上硅谷大数据项目之一Flink
(作者:尚硅谷大数据研发部)

版本：V1.12.0

Flink简介 初识Flink 

Flink起源于Stratosphere项目，Stratosphere是在2010~2014年4月Stratosphere代码被复制并捐赠给Apache软件基金会，参化项目的初始成员是Stratosphere2014年12月，系统核心开发人员，Flink一跃成为Apache顶级软件基金会项目。
在德语中，Flink一个词表示快速和灵巧，项目使用松鼠的颜色图案作为logo，这不仅是因为松鼠快速灵巧，还因为柏林松鼠有迷人的红棕色Flink的松鼠logo尾巴可爱，尾巴的颜色和Apache软件基金会logo颜色相呼应，也就是说，这是一个Apache风格的松鼠。

Flink项目的概念是：Apache Flink开源流处理框架是为分布式、高性能、随时可用、准确的流处理应用程序而建立的。
Apache Flink用于有状态计算无界和有界数据流的框架和分布式处理引擎。Flink以内存执行速度和任群环境中运行，计算内存执行速度和任何规模。

Flink的重要特点 事件驱动型(Event-driven) 

事件驱动应用程序是一种状态应用程序，它从一个或多个事件流中提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。更典型的是kafka几乎所有以事件为代表的新闻队列都是驱动应用。(Flink计算也是事件驱动型)
不同的是SparkStreaming如图所示：

事件驱动型:  流与批的世界观 

批处理的特点是有界、大量，非常适合需要访问全套记录才能完成的计算工作，一般用于离线统计。
流处理的特点是无界、实时、 对于通过系统传输的每个数据项，通常用于实时统计，而不是执行整个数据集。
在spark在世界观中，一切都是由批次组成的，离线数据是大批次，实时数据是由无限小批次组成的。
而在flink在世界观中，一切都是由流组成的，离线数据是有界限的流，实时数据是无界限的流，这就是所谓的有界流和无界流。

无限数据流：
无界数据流有一个开始，但还没有结束。它们不会在生成时终止并提供数据，必须连续处理无界流，即获得后必须立即处理event。对于无限数据流，我们不能等待所有数据到达，因为输入是无限的，不会在任何时候完成。处理无界数据通常需要以特定的顺序(如事件的顺序)获取event，为了推断结果的完整性。
有界数据流：
有界数据流有一个明确的开始和结束定义。在执行任何计算之前，都可以通过获取所有数据来处理有界流。处理有界流不需要有序获取，因为它可以始终对有界数据集进行排序，有界流处理也称为批处理。
分层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优化批处理数据.
Spark or Flink
Spark 和 Flink 一开始，他们都有同样的梦想。他们都希望用同样的技术统一流程处理和批处理，但他们走了两条完全不同的道路。前者以批处理技术为基础，试图支持批处理上的流程计算；后者认为流程计算技术是最基本的，在流程计算的基础上支持批处理。由于这种结构的不同，两者在未来能做的事情上会有一些细微的差异。例如，在低延迟场景中，Spark 基于微批处理的方式需要同步，会有额外的费用，所以延迟不能达到极致。在大数据处理的低延迟场景中，Flink 有很大的优势。
Spark和Flink主要区别在于计算模型不同。Spark微批处理模型采用，Flink采用基于操作符的连续流模型。因此，对Apache Spark和Apache Flink其实选择已经成为计算模型的选择，需要在延迟、吞吐量吐量和可靠性。
如果企业必须从技术选择中选择Spark和Flink我们推荐使用这两个主流框架中的一个进行流数据处理Flink，主(显而易见)要(易见)的原因是:
Flink灵活的窗口
Exactly Once语义保证
事件时间(event-time)语义(处理乱序数据或延迟数据)
这两个原因可以大大解放程序员， 加快编程效率， 把原本需要程序员手动完成的工作交给框架，太棒了，赞不绝口！

Flink的应用 应用Flink的场景 

事件驱动应用
事件驱动应用程序是一种状态应用程序，它从一个或多个事件流中提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。
事件驱动应用是在计算存储分离的传统应用的基础上进化而来的。在传统架构中，应用程序需要读写远程事务数据库。
相反，事件驱动应用程序是基于状态流处理。在设计中，数据和计算不会分开，应用程序只需访问本地（内存或磁盘）即可获取数据。系统容错的实现取决于定期写入远程持久存储 checkpoint。下图描述了传统应用与事件驱动应用架构的区别。

事件驱动型应用无须查询远程数据库，本地数据访问使得它具有更高的吞吐和更低的延迟。由于定期存储到远程持久存储， checkpoint 工作可以异步增量完成，对正常事件的处理影响不大。事件驱动应用的优势不限于本地数据访问。在传统的分层架构下，多个应用程序通常共享相同的数据库，因此需要仔细协调数据库本身的任何变化（如应用程序更新或服务扩展）。另一方面，由于事件驱动应用只需要考虑自己的数据，因此在更改数据或扩大服务容量时所需的协调工作将大大减少。
数据分析应用
Apache Flink 支持流式和批量分析应用。

Flink 为持续流式分析和批量分析提供了良好的支持。具体来说，它内置符合要求 ANSI 标准的 SQL 接口统一批流查询的语义。无论是记录事件的静态数据集还是实时事件流，都是一样的 SQL 查询会得到一致的结果。同时 Flink 它还支持丰富的用户自定义函数 SQL 执行定制代码。如果需要进一步定制逻辑，可以使用 Flink DataStream API 和 DataSet API 控制水平较低。此外，Flink 的 Gelly 库为基于批量数据集的大规模高性能图分析提供算法和构建模块支持。
数据管道应用
提取-转换-加载（ETL）是存储系统之间数据转换和迁移的常用方法。ETL 操作通常会定期触发，将数据从事务数据库复制到分析数据库或数据仓库。
数据管道和 ETL 所有类似的操作都可以转换和丰富数据，并将其从存储系统移动到另一个。但数据管道以连续流模式运行，而不是周期性触发。因此，它支持从不断生成的数据源读取记录，并以低延迟移动到终点。例如，数据管可以用来监控文件系统目录中的新文件，并将其数据写入事件日志；另一个应用程序可以将事件流物化到数据库或构建和优化查询索引。
下图描述了周期性 ETL 操作与连续数据管道的差异。

应用Flink的行业 

电子商务和营销
数据报告，广告
物联网（IOT）
传感器实时数据采集显示、实时报警、交通运输行业
物流配送与服务业
单状态实时更新、通知信息推送、电信业基站流量调配
银行和金融业
实时结算和通知推送，实时检测异常行为
应用Flink的企业

Flink快速上手
创建maven项目
POM文件中添加需要的依赖:

|

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

批处理WordCount

// 1. 创建执行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
// 2. 从文件读取数据 按行读取(存储的元素就是每行的文本)
DataSource lineDS = env.readTextFile(“input/words.txt”);
// 3. 转换数据格式
FlatMapOperator> wordAndOne = lineDS
.flatMap((String line, Collector> 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> wordAndOneUG = wordAndOne.groupBy(0);
// 5. 分组内聚合统计
AggregateOperator> sum = wordAndOneUG.sum(1);

// 6. 打印结果
sum.print();
流处理WordCount
有界流
// 1. 创建流式执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 读取文件
DataStreamSource lineDSS = env.readTextFile(“input/words.txt”);
// 3. 转换数据格式
SingleOutputStreamOperator> wordAndOne = lineDSS
.flatMap((String line, Collector 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, String> wordAndOneKS = wordAndOne
.keyBy(t -> t.f0);
// 5. 求和
Sing leOutputStreamOperator> result = wordAndOneKS
.sum(1);
// 6. 打印
result.print();
// 7. 执行
env.execute();
无界流
// 1. 创建流式执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 读取文件
DataStreamSource lineDSS = env.socketTextStream(“hadoop162”, 9999);
// 3. 转换数据格式
SingleOutputStreamOperator> wordAndOne = lineDSS
.flatMap((String line, Collector 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, String> wordAndOneKS = wordAndOne
.keyBy(t -> t.f0);
// 5. 求和
SingleOutputStreamOperator> result = wordAndOneKS
.sum(1);
// 6. 打印
result.print();
// 7. 执行
env.execute();
Flink部署
开发模式
咱们前面在idea中运行Flink程序的方式就是开发模式.
local-cluster模式
Flink中的Local-cluster(本地集群)模式,主要用于测试, 学习.
local-cluster模式配置
local-cluster模式基本属于零配置.
上传Flink的安装包flink-1.12.0-bin-scala_2.11.tgz到hadoop162
解压
tar -zxvf flink-1.12.0-bin-scala_2.11.tgz -C /opt/module
进入目录/opt/module, 复制flink-local
cd /opt/module
cp -r flink-1.12.0 flink-local
在local-cluster模式下运行无界的WordCount
打包idea中的应用
把不带依赖的jar包上传到目录/opt/module/flink-local下
启动本地集群
bin/start-cluster.sh
在hadoop162中启动netcat
nc -lk 9999
注意: 如果没有安装netcat需要先安装:
sudo yum install -y nc
命令行提交Flink应用
bin/flink run -m hadoop162:8081 -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
在浏览器中查看应用执行情况
http://hadoop162:8081

也可以在log日志查看执行结果

cat flink-atguigu-taskexecutor-0-hadoop162.out

也可以在WEB UI提交应用


Standalone模式

Standalone模式又叫独立集群模式.
Standalone模式配置
复制flink-standalone
cp -r flink-1.12.0 flink-standalone
修改配置文件:flink-conf.yaml
jobmanager.rpc.address: hadoop162
修改配置文件:workers
hadoop163
hadoop164
分发flink-standalone到其他节点
Standalone模式运行无界流WorkCount
启动standalone集群
bin/start-cluster.sh
命令行提交Flink应用
bin/flink run -m hadoop162:8081 -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
查看执行情况与本地集群一致.
也支持Web UI界面提交Flink应用
Standalone高可用(HA)
任何时候都有一个 主 JobManager 和多个备用 JobManagers，以便在主节点失败时有备用 JobManagers 来接管集群。这保证了没有单点故障，一旦备 JobManager 接管集群，作业就可以正常运行。主备 JobManager 实例之间没有明显的区别。每个 JobManager 都可以充当主备节点。

修改配置文件: 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
修改配置文件: masters
hadoop162:8081
hadoop163:8081
分发修改的后配置文件到其他节点
在/etc/profile.d/my.sh中配置环境变量
export HADOOP_CLASSPATH=hadoop classpath
注意:
需要提前保证HAOOP_HOME环境变量配置成功
分发到其他节点
首先启动dfs集群和zookeeper集群
启动standalone HA集群
bin/start-cluster.sh
可以分别访问
http://hadoop162:8081
http://hadoop163:8081
在zkCli.sh中查看谁是leader
get /flink-standalone/cluster_atguigu/leader/rest_server_lock

杀死hadoop162上的Jobmanager, 再看leader

注意: 不管是不是leader从WEB UI上看不到区别, 并且都可以与之提交应用.
Yarn模式
独立部署（Standalone）模式由Flink自身提供计算资源，无需其他框架提供资源，这种方式降低了和其他第三方资源框架的耦合性，独立性非常强。但是你也要记住，Flink主要是计算框架，而不是资源调度框架，所以本身提供的资源调度并不是它的强项，所以还是和其他专业的资源调度框架集成更靠谱，所以接下来我们来学习在强大的Yarn环境中Flink是如何使用的。（其实是因为在国内工作中，Yarn使用的非常多）
把Flink应用提交给Yarn的ResourceManager, Yarn的ResourceManager会申请容器从Yarn的NodeManager上面. Flink会创建JobManager和TaskManager在这些容器上.Flink会根据运行在JobManger上的job的需要的slot的数量动态的分配TaskManager资源
Yarn模式配置
复制flink-yarn
cp -r flink-1.11.2 flink-yarn
配置环境变量HADOOP_CLASSPATH, 如果前面已经配置可以忽略.
在/etc/profile.d/my.sh中配置
export HADOOP_CLASSPATH=hadoop classpath
Yarn运行无界流WordCount
启动hadoop集群(hdfs, yarn)
运行无界流
bin/flink run -t yarn-per-job -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
在yarn的ResourceManager界面查看执行情况

Flink on Yarn的3种部署模式

Flink提供了yarn上运行的3模式，分别为Application Mode, 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函数是在集群中执行的
官方建议:
出于生产的需求, 我们建议使用Per-job or Application Mode,因为他们给应用提供了更好的隔离!

Per-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
Session-Cluster模式执行无界流WordCount
启动一个Flink-Session
bin/yarn-session.sh -d
在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应用
Application 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
Yarn模式高可用
Yarn模式的高可用和Standalone模式的高可用原理不一样.
Standalone模式中, 同时启动多个Jobmanager, 一个为leader其他为standby, 当leader挂了, 其他的才会有一个成为leader.
yarn的高可用是同时只启动一个Jobmanager, 当这个Jobmanager挂了之后, yarn会再次启动一个, 其实是利用的yarn的重试次数来实现的高可用.
在yarn-site.xml中配置

yarn.resourcemanager.am.max-attempts
4

The maximum number of application master execution attempts.


注意: 配置完不要忘记分发, 和重启yarn
在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
启动yarn-session
杀死Jobmanager, 查看他的复活情况
注意: yarn-site.xml中是它活的次数的上限, flink-conf.xml中的次数应该小于这个值.
Scala REPL
scala 交互环境.
local模式启动 REPL
/opt/module/flink-local » bin/start-scala-shell.sh local
yarn-session 模式启动
先启动一个yarn-session, 然后就可以把shell跑在yarn-session上了
bin/start-scala-shell.sh yarn
K8S & Mesos模式
Mesos是Apache下的开源分布式资源管理框架，它被称为是分布式系统的内核,在Twitter得到广泛使用,管理着Twitter超过30,0000台服务器上的应用部署，但是在国内，依然使用着传统的Hadoop大数据框架，所以国内使用mesos框架的并不多，这里我们就不做过多讲解了。
容器化部署时目前业界很流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes（k8s），而Flink也在最近的版本中支持了k8s部署模式。这里我们也不做过多的讲解.
Flink运行架构
运行架构
https://ci.apache.org/projects/flink/flink-docs-release-1.11/fig/processes.svg

Flink运行时包含2种进程:1个JobManager和至少1个TaskManager
客户端
严格上说, 客户端不是运行和程序执行的一部分, 而是用于准备和发送dataflow到JobManager. 然后客户端可以断开与JobManager的连接(detached mode), 也可以继续保持与JobManager的连接(attached mode)
客户端作为触发执行的java或者scala代码的一部分运行, 也可以在命令行运行:bin/flink run …
JobManager
控制一个应用程序执行的主进程，也就是说，每个应用程序都会被一个的JobManager所控制执行。
JobManager会先接收到要执行的应用程序，这个应用程序会包括：作业图（JobGraph）、逻辑数据流图（logical dataflow graph）和打包了所有的类、库和其它资源的JAR包。
JobManager会把JobGraph转换成一个物理层面的数据流图，这个图被叫做“执行图”（ExecutionGraph），包含了所有可以并发执行的任务。JobManager会向资源管理器（ResourceManager）请求执行任务必要的资源，也就是任务管理器（TaskManager）上的插槽（slot）。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的TaskManager上。
而在运行过程中，JobManager会负责所有需要中央协调的操作，比如说检查点（checkpoints）的协调。
这个进程包含3个不同的组件
ResourceManager
负责资源的管理，在整个 Flink 集群中只有一个 ResourceManager. 注意这个ResourceManager不是Yarn中的ResourceManager, 是Flink中内置的, 只是赶巧重名了而已.
主要负责管理任务管理器（TaskManager）的插槽（slot），TaskManger插槽是Flink中定义的处理资源单元。
当JobManager申请插槽资源时，ResourceManager会将有空闲插槽的TaskManager分配给JobManager。如果ResourceManager没有足够的插槽来满足JobManager的请求，它还可以向资源提供平台发起会话，以提供启动TaskManager进程的容器。另外，ResourceManager还负责终止空闲的TaskManager，释放计算资源。
Dispatcher
负责接收用户提供的作业，并且负责为这个新提交的作业启动一个新的JobManager 组件. Dispatcher也会启动一个Web UI，用来方便地展示和监控作业执行的信息。Dispatcher在架构中可能并不是必需的，这取决于应用提交运行的方式。
JobMaster
JobMaster负责管理单个JobGraph的执行.多个Job可以同时运行在一个Flink集群中, 每个Job都有一个自己的JobMaster.
TaskManager
Flink中的工作进程。通常在Flink中会有多个TaskManager运行，每一个TaskManager都包含了一定数量的插槽（slots）。插槽的数量限制了TaskManager能够执行的任务数量。
启动之后，TaskManager会向资源管理器注册它的插槽；收到资源管理器的指令后，TaskManager就会将一个或者多个插槽提供给JobManager调用。JobManager就可以向插槽分配任务（tasks）来执行了。
在执行过程中，一个TaskManager可以跟其它运行同一应用程序的TaskManager交换数据。
核心概念
TaskManager与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的受管理的内存。
Parallelism（并行度）

一个特定算子的子任务（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中的宽依赖
Task与SubTask
一个算子就是一个Task. 一个算子的并行度是几, 这个Task就有几个SubTask
Operator Chains（任务链）
相同并行度的one to one操作，Flink将这样相连的算子链接在一起形成一个task，原来的算子成为里面的一部分。 每个task被一个线程执行.
将算子链接成task是非常有效的优化：它能减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。链接的行为可以在编程API中进行指定。

ExecutionGraph（执行图）

由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();

提交流程
高级视角提交流程(通用提交流程)

我们来看看当一个应用提交执行时，Flink的各个组件是如何交互协作的：

yarn-cluster提交流程per-job

Flink任务提交后，Client向HDFS上传Flink的Jar包和配置
向Yarn ResourceManager提交任务，ResourceManager分配Container资源
通知对应的NodeManager启动ApplicationMaster，ApplicationMaster启动后加载Flink的Jar包和配置构建环境，然后启动JobManager
ApplicationMaster向ResourceManager申请资源启动TaskManager
ResourceManager分配Container资源后，由ApplicationMaster通知资源所在节点的NodeManager启动TaskManager
NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager
TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务。
Flink流处理核心编程

和其他所有的计算框架一样，Llink也有一些基础的开发步骤以及基础，核心的API，从开发步骤的角度来讲，主要分为四大部分

Environment

Flink Job在提交执行计算时，需要首先建立和Flink框架之间的联系，也就指的是当前的flink运行环境，只有获取了环境信息，才能将task调度到不同的taskManager执行。而这个环境对象的获取方式相对比较简单
// 批处理环境
ExecutionEnvironment benv = ExecutionEnvironment.getExecutionEnvironment();
// 流式数据处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Source

Flink框架可以从不同的来源获取数据，将数据提交给框架进行处理, 我们将获取数据的来源称之为数据源(Source)。
准备工作
导入注解工具依赖, 方便生产POJO类

org.projectlombok lombok 1.18.16 provided 准备一个WaterSensor类方便演示 package com.atguigu.flink.java.chapter_5;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**

• @Author lzc
• @Date 2020/12/8 22:22
• 水位传感器：用于接收水位数据
• id:传感器编号
• ts:时间戳
• vc:水位
  */
  @Data
  @NoArgsConstructor
  @AllArgsConstructor
  public class WaterSensor {
  private String id;
  private Long ts;
  private Integer vc;
  }
  从Java的集合中读取数据
  一般情况下，可以将数据临时存储到内存中，形成特殊的数据结构后，作为数据源使用。这里的数据结构采用集合类型是比较普遍的。
  package com.atguigu.flink.java.chapter_5;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/8 22:44
  */
  public class Flink01_Source_Collection {
  public static void main(String[] args) throws Exception {
  List 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();
  

  }
  }
  从文件读取数据
  package com.atguigu.flink.java.chapter_5;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/8 22:44
  */
  public class Flink02_Source_File {
  public static void main(String[] args) throws Exception {

   // 1. 创建执行环境
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   env
     .readTextFile("input")
     .print();
  
   env.execute();
  

  }
  }
  说明:
  参数可以是目录也可以是文件
  路径可以是相对路径也可以是绝对路径
  相对路径是从系统属性user.dir获取路径: idea下是project的根目录, standalone模式下是集群节点根目录
  也可以从hdfs目录下读取, 使用路径:hdfs://…, 由于Flink没有提供hadoop相关依赖, 需要pom中添加相关依赖:
  
  org.apache.hadoop
  hadoop-client
  3.1.3
  provided
  
  从Socket读取数据
  参考第1章无界流读取
  从Kafka读取数据
  添加相应的依赖
  
  org.apache.flink
  flink-connector-kafka_2.11
  1.11.2
  
  参考代码
  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;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/8 22:44
  */
  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();
  

  }
  }
  自定义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;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/8 22:44
  */
  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 {
  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 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
  */
  Transform

转换算子可以把一个或多个DataStream转成一个新的DataStream.程序可以把多个复杂的转换组合成复杂的数据流拓扑.
map
作用
将数据流中的数据进行转换, 形成新的数据流，消费一个元素并产出一个元素

参数
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;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/10 7:17
  */
  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() {
         @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;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/10 7:17
  */
  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 {

   @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;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/10 7:17
  */
  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 {
  // 默认生命周期方法, 初始化方法, 在每个并行度上只会被调用一次
  @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;
   }
  

  }
  }

  默认生命周期方法, 初始化方法, 在每个并行度上只会被调用一次, 而且先被调用
  默认生命周期方法, 最后一个方法, 做一些清理工作, 在每个并行度上只调用一次, 而且是最后被调用
  getRuntimeContext()方法提供了函数的RuntimeContext的一些信息，例如函数执行的并行度，任务的名字，以及state状态. 开发人员在需要的时候自行调用获取运行时上下文对象.
  flatMap
  作用
  消费一个元素并产生零个或多个元素

参数
FlatMapFunction实现类
返回
DataStream → DataStream
示例
匿名内部类写法
// 新的流存储每个元素的平方和3次方
env
.fromElements(1, 2, 3, 4, 5)
.flatMap(new FlatMapFunction() {
@Override
public void flatMap(Integer value, Collector 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 out) -> {
out.collect(value * value);
out.collect(value * value * value);
}).returns(Types.INT)
.print();
说明: 在使用Lambda表达式表达式的时候, 由于泛型擦除的存在, 在运行的时候无法获取泛型的具体类型, 全部当做Object来处理, 及其低效, 所以Flink要求当参数中有泛型的时候, 必须明确指定泛型的类型.
filter
作用
根据指定的规则将满足条件（true）的数据保留，不满足条件(false)的数据丢弃

参数
FlatMapFunction实现类
返回
DataStream → DataStream
示例
匿名内部类写法
// 保留偶数, 舍弃奇数
env
.fromElements(10, 3, 5, 9, 20, 8)
.filter(new FilterFunction() {
@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();
keyBy
作用
把流中的数据分到不同的分区中.具有相同key的元素会分到同一个分区中.一个分区中可以有多重不同的key.
在内部是使用的hash分区来实现的.

参数
Key选择器函数: interface KeySelector
注意: 什么值不可以作为KeySelector的Key:
没有覆写hashCode方法的POJO, 而是依赖Object的hashCode. 因为这样分组没有任何的意义: 每个元素都会得到一个独立无二的组. 实际情况是:可以运行, 但是分的组没有意义.
任何类型的数组
返回
DataStream → KeyedStream
示例
匿名内部类写法
// 奇数分一组, 偶数分一组
env
.fromElements(10, 3, 5, 9, 20, 8)
.keyBy(new KeySelector() {
@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();
shuffle
作用
把流中的元素随机打乱. 对同一个组数据, 每次执行得到的结果都不同.

参数
无
返回
DataStream → DataStream
示例
env
.fromElements(10, 3, 5, 9, 20, 8)
.shuffle()
.print();
env.execute();
split和select
已经过时, 在1.12中已经被移除
作用
在某些情况下，我们需要将数据流根据某些特征拆分成两个或者多个数据流，给不同数据流增加标记以便于从流中取出.
split用于给流中的每个元素添加标记. select用于根据标记取出对应的元素, 组成新的流.

参数
split参数: interface OutputSelector
select参数: 字符串
返回
split: SingleOutputStreamOperator -> SplitStream
slect: SplitStream -> DataStream
示例
匿名内部类写法
// 奇数一个流, 偶数一个流
SplitStream splitStream = env
.fromElements(10, 3, 5, 9, 20, 8)
.split(new OutputSelector() {
@Override
public Iterable select(Integer value) {
return value % 2 == 0
? Collections.singletonList(“偶数”)
: Collections.singletonList(“奇数”);
}
});
splitStream
.select(“偶数”)
.print(“偶数”);

splitStream
.select(“奇数”)
.print(“奇数”);
env.execute();
Lambda表达式写法
// 奇数一个流, 偶数一个流
SplitStream splitStream = env
.fromElements(10, 3, 5, 9, 20, 8)
.split(value -> value % 2 == 0
? Collections.singletonList(“偶数”)
: Collections.singletonList(“奇数”));
splitStream
.select(“偶数”)
.print(“偶数”);

splitStream
.select(“奇数”)
.print(“奇数”);
env.execute();
connect
作用
在某些情况下，我们需要将两个不同来源的数据流进行连接，实现数据匹配，比如订单支付和第三方交易信息，这两个信息的数据就来自于不同数据源，连接后，将订单支付和第三方交易信息进行对账，此时，才能算真正的支付完成。
Flink中的connect算子可以连接两个保持他们类型的数据流，两个数据流被connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

参数
另外一个流
返回
DataStream[A], DataStream[B] -> ConnectedStreams[A,B]
示例
DataStreamSource intStream = env.fromElements(1, 2, 3, 4, 5);
DataStreamSource stringStream = env.fromElements(“a”, “b”, “c”);
// 把两个流连接在一起: 貌合神离
ConnectedStreams cs = intStream.connect(stringStream);
cs.getFirstInput().print(“first”);
cs.getSecondInput().print(“second”);
env.execute();
注意:
两个流中存储的数据类型可以不同
只是机械的合并在一起, 内部仍然是分离的2个流
只能2个流进行connect, 不能有第3个参与
union
作用
对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream

示例
DataStreamSource stream1 = env.fromElements(1, 2, 3, 4, 5);
DataStreamSource stream2 = env.fromElements(10, 20, 30, 40, 50);
DataStreamSource stream3 = env.fromElements(100, 200, 300, 400, 500);

// 把多个流union在一起成为一个流, 这些流中存储的数据类型必须一样: 水乳交融
stream1
.union(stream2)
.union(stream3)
.print();
connect与 union 区别：
union之前两个或多个流的类型必须是一样，connect可以不一样
connect只能操作两个流，union可以操作多个。
简单滚动聚合算子
常见的滚动聚合算子
sum, min,max
minBy,maxBy
作用
KeyedStream的每一个支流做聚合。执行完成后，会将聚合的结果合成一个流返回，所以结果都是DataStream
参数
如果流中存储的是POJO或者scala的样例类, 参数使用字段名. 如果流中存储的是元组, 参数就是位置(基于0…)
返回
KeyedStream -> SingleOutputStreamOperator
示例
示例1
DataStreamSource stream = env.fromElements(1, 2, 3, 4, 5);
KeyedStream kbStream = stream.keyBy(ele -> ele % 2 == 0 ? “奇数” : “偶数”);
kbStream.sum(0).print(“sum”);
kbStream.max(0).print(“max”);
kbStream.min(0).print(“min”);
示例2
ArrayList 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 kbStream = env
.fromCollection(waterSensors)
.keyBy(WaterSensor::getId);

kbStream
.sum(“vc”)
.print(“maxBy…”);

注意:
分组聚合后, 理论上只能取分组字段和聚合结果, 但是Flink允许其他的字段也可以取出来, 其他字段默认情况是取的是这个组内第一个元素的字段值
示例3:
ArrayList 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 kbStream = env
.fromCollection(waterSensors)
.keyBy(WaterSensor::getId);

kbStream
.maxBy(“vc”, false)
.print(“max…”);

env.execute();
注意:
maxBy和minBy可以指定当出现相同值的时候,其他字段是否取第一个. true表示取第一个, false表示取与最大值(最小值)同一行的.
reduce
作用
一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。
为什么还要把中间值也保存下来? 考虑流式数据的特点: 没有终点, 也就没有最终的概念了. 任何一个中间的聚合结果都是值!
参数
interface ReduceFunction
返回
KeyedStream -> SingleOutputStreamOperator
示例
匿名内部类写法
ArrayList 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 kbStream = env
.fromCollection(waterSensors)
.keyBy(WaterSensor::getId);

kbStream
.reduce(new ReduceFunction() {
@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…”);
注意:
聚合后结果的类型, 必须和原来流中元素的类型保持一致!
process
作用
process算子在Flink算是一个比较底层的算子, 很多类型的流上都可以调用, 可以从流中获取更多的信息(不仅仅数据本身)
示例1: 在keyBy之前的流上使用
env
.fromCollection(waterSensors)
.process(new ProcessFunction>() {
@Override
public void processElement(WaterSensor value,
Context ctx,
Collector> out) throws Exception {
out.collect(new Tuple2<>(value.getId(), value.getVc()));
}
})
.print();

示例2: 在keyBy之后的流上使用
env
.fromCollection(waterSensors)
.keyBy(WaterSensor::getId)
.process(new KeyedProcessFunction>() {
@Override
public void processElement(WaterSensor value, Context ctx, Collector> out) throws Exception {
out.collect(new Tuple2<>(“key是:” + ctx.getCurrentKey(), value.getVc()));
}
})
.print();
对流重新分区的几个算子
KeyBy
先按照key分组, 按照key的双重hash来选择后面的分区
shuffle
对流中的元素随机分区

reblance
对流中的元素平均分布到每个区.当处理倾斜数据的时候, 进行性能优化
rescale
同 rebalance一样, 也是平均循环的分布数据. 但是要比rebalance更高效, 因为rescale不需要通过网络, 完全走的"管道"

Sink

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需要用户自定义!

KafkaSink

添加Kafka Connector依赖

org.apache.flink
flink-connector-kafka_2.11
1.11.2


com.alibaba
fastjson
1.2.75

启动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;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/10 14:46
  */
  public class Flink01_Sink_Kafka {
  public static void main(String[] args) throws Exception {
  ArrayList 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("hadoop102:9092", "topic_sensor", new SimpleStringSchema()));
  
  
   env.execute();
  

  }
  }
  在linux启动一个消费者, 查看是否收到数据
  bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_sensor

  RedisSink
  添加Redis Connector依赖

org.apache.flink flink-connector-redis_2.11 1.1.5 启动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;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/10 14:46
  */
  public class Flink02_Sink_Redis {
  public static void main(String[] args) throws Exception {
  ArrayList 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() {
         /*
           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的重复了, 后面的会把前面的覆盖掉
ElasticsearchSink
添加Elasticsearch Connector依赖

org.apache.flink flink-connector-elasticsearch6_2.11 1.12.0 启动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;

/**

• @Author lizhenchao@atguigu.cn

• @Date 2020/12/10 14:46
  */
  public class Flink03_Sink_ES {
  public static void main(String[] args) throws Exception {
  ArrayList 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 esHosts = Arrays.asList(
     new HttpHost("hadoop102", 9200),
     new HttpHost("hadoop103", 9200),
     n