四、Flink运行架构_flink一个中存在几个进程

4 Flink运行架构

4.1 运行架构

https://ci.apache.org/projects/flink/flink-docs-release-1.11/fig/processes.svg
1

Flink运行时包含2种进程:1个JobManager和至少1个TaskManager

4.1.1 客户端

严格上说, 客户端不是运行和程序执行的一部分, 而是用于准备和发送dataflow到JobManager. 然后客户端可以断开与JobManager的连接(detached mode), 也可以继续保持与JobManager的连接(attached mode)

客户端作为触发执行的java或者scala代码的一部分运行, 也可以在命令行运行:

bin/flink run ...
1

4.1.1.1 JobManager

控制一个应用程序执行的主进程，也就是说，每个应用程序都会被一个不同的JobManager所控制执行。

JobManager会先接收到要执行的应用程序，这个应用程序会包括：作业图（JobGraph）、逻辑数据流图（logical dataflow graph）和打包了所有的类、库和其它资源的JAR包。

JobManager会把JobGraph转换成一个物理层面的数据流图，这个图被叫做“执行图”（ExecutionGraph），包含了所有可以并发执行的任务。JobManager会向资源管理器（ResourceManager）请求执行任务必要的资源，也就是任务管理器（TaskManager）上的插槽（slot）。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的TaskManager上。

而在运行过程中，JobManager会负责所有需要中央协调的操作，比如说检查点（checkpoints）的协调。

这个进程包含3个不同的组件

JobManager主要作用：

1. 接受客户端请求
2. 划分任务
3. 申请资源

4.1.2.2 ResourceManager

负责资源的管理，在整个 Flink 集群中只有一个 ResourceManager. 注意这个ResourceManager 不是 Yarn 中的 ResourceManager, 是 Flink 中内置的, 只是赶巧重名了而已.

主要负责管理任务管理器（TaskManager）的插槽（slot），TaskManger 插槽是 Flink 中定义的处理资源单元。

当 JobManager 申请插槽资源时，ResourceManager 会将有空闲插槽的 TaskManager 分配给 JobManager。如果 ResourceManager 没有足够的插槽来满足 JobManager 的请求，它还可以向资源提供平台发起会话，以提供启动 TaskManager 进程的容器。另外，ResourceManager 还负责终止空闲的 TaskManager，释放计算资源。

4.1.2.3 Dispatcher

负责接收用户提供的作业，并且负责为这个新提交的作业启动一个新的 JobMaster 组件. Dispatcher 也会启动一个 Web UI，用来方便地展示和监控作业执行的信息。Dispatcher 在架构中可能并不是必需的，这取决于应用提交运行的方式。

4.1.2.4 JobMaster

JobMaster 负责管理单个 JobGraph 的执行.多个 Job 可以同时运行在一个 Flink 集群中, 每个 Job 都有一个自己的 JobMaster。

4.1.3 TaskManager

Flink 中的工作进程。通常在 Flink 中会有多个 TaskManager运行。每一个 TaskManager 都包含了一定数量的插槽（slots）。插槽的数量限制了 TaskManager 能够执行的任务数量。

启动之后，TaskManager 会向资源管理器注册它的插槽；收到资源管理器的指令后，TaskManager 就会将一个或者多个插槽提供给 JobManager 调用。JobManager 就可以向插槽分配任务（tasks）来执行了。

在执行过程中，一个 TaskManager 可以跟其它运行同一应用程序的 TaskManager 交换数据。

总结：

1. 工作进程，任务都在 TaskManager 上运行
2. TaskManager 中有资源（Slot）
3. 需要向 JobManager 进行交互从而进行资源的注册和使用
4. 多个 TaskManager 可以交换数据

4.2 核心概念

4.2.1 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 的受管理的内存。

总结：

1. slot是可以共享的（Job内部），外部共享只有一种情况（Session）。
2. slot会均分内存资源，进而达到内存隔离，相互之间不会占用内存。但cpu资源不会隔离，可以共享cpu资源。

4.2.2 Parallelism(并行度)

一个特定算子的子任务（subtask）的个数被称之为这个算子的并行度（parallelism），一般情况下，一个流程序的并行度，可以认为就是其所有算子中最大的并行度。一个程序中，不同的算子可能具有不同的并行度，如果是yarn模式会动态申请资源slot个数小于并行度那么程序将会卡死一直处于create状态，等待足够资源，才运行**，如果是standalone模式，不会动态申请资源**，slot个数小于并行度那么程序将会直接失败。

并行度优先级：

算子指定 > env全局指定 > 提交参数 > 配置文件

4.2.3 Task 与 SubTask

不同算子的子任务（SubTask）经过一定的优化，串在一起，形成一个新的 SubTask 对 TaskManager 来讲，就是一个 Task。

sum算子  print算子  并行度都是 3
  ⭕		  ⭕
  ⭕		  ⭕
  ⭕		  ⭕

sum算子 跟 print算子 满足某种不可描述的关系，可以串在一起
（⭕  ⭕） -> 新的 subtask -> 对 TaskManager来讲，就是一个 Task
（⭕  ⭕） -> 新的 subtask -> 对 TaskManager来讲，就是一个 Task
（⭕  ⭕） -> 新的 subtask -> 对 TaskManager来讲，就是一个 Task
1
2
3
4
5
6
7
8
9

4.2.4 Operator Chains(任务链)

Stream在算子之间传输数据的形式可以是 one-to-one(forwarding) 的模式也可以是 redistributing 的模式，具体是哪一种形式，取决于算子的种类。

One-to-one:

stream(比如在 source 和 flatmap operator之间)维护着分区以及元素的顺序。那意味着 flatmap 算子的子任务看到的元素的个数以及顺序跟 source 算子的子任务生产的元素的个数、顺序相同，map、fliter、flatMap 等算子都是 one-to-one 的对应关系。

Redistributing:

stream(map()跟 keyBy/window 之间或者 keyBy/window 跟 sink 之间)的分区会发生改变。每一个算子的子任务依据所选择的 transformation 发送数据到不同的目标任务。例如，keyBy()基于 hashCode 重分区、broadcast 和 rebalance 会随机重新分区，这些算子都会引起 redistribute 过程，而 redistribute 过程就类似于 Spark 中的 shuffle 过程。

相同并行度的 one to one 操作，Flink 将这样相连的算子链接在一起形成一个 task，原来的算子成为里面的一部分。 每个 task 被一个线程执行。

将算子链接成 task 是非常有效的优化：它能减少线程之间的切换和基于缓存区的数据交换，减少时延的同时提升吞吐量。链接的行为可以在编程API中进行指定。

算子.startNewChain() => 与前面都断开

算子.disableChaining() => 与前后都断开

env.disableOperatorChaining() => 全局都不串
1
2
3
4
5

断开操作链的好处在于减少某个slot的压力

共享组: 

.slotSharingGroup("group1")
1
2
3

设置共享组后，slot 的使用个数并不仅仅是与程序中最大算子的并行度有关，当算子设置共享组后会使用额外的 slot 来执行，默认只有一个共享组，因此在默认只有一个共享组的情况下，slot 的个数等于程序中最大算子的并行度。

任务链以及共享组：

任务链的好处：避免数据跨节点传输，减少线程间的切换。

断开任务链的好处：减少某个slot的压力。

默认情况下所有算子都是同一个共享组，任务所需要的slot数量：最大算子的并行度。

当使用共享组时，任务所需要的slot数量：每个共享组中最大并行度的和。

4.2.5 ExecutionGraph(执行图)

由 Flink 程序直接映射成的数据流图是 StreamGraph，也被称为逻辑流图，因为它们表示的是计算逻辑的高级视图。为了执行一个流处理程序，Flink 需要将逻辑流图转换为物理数据流图（也叫执行图），详细说明程序的执行方式。

Flink 中的执行图可以分成四层：

StreamGraph -> JobGraph -> ExecutionGraph -> PhysicalGraph

StreamGraph：是根据用户通过 Stream API 编写的代码生成的最初的图。用来表示程序的拓扑结构。

JobGraph：StreamGraph经过优化后生成了 JobGraph，是提交给 JobManager 的数据结构。主要的优化为: 将多个符合条件的节点 chain 在一起作为一个节点，这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。

ExecutionGraph：JobManager 根据 JobGraph 生成 ExecutionGraph。ExecutionGraph 是 JobGraph 的并行化版本，是调度层最核心的数据结构。

**PhysicalGraph：**JobManager 根据 ExecutionGraph 对 Job 进行调度后，在各个 TaskManager 上部署 Task 后形成的“图”，并不是一个具体的数据结构。

2个并发度（Source为1个并发度）的 SocketTextStreamWordCount 四层执行图的演变过程。

	env.socketTextStream()
	.flatMap(…)
	.keyBy(0)
	.sum(1)
	.print();
1
2
3
4
5

4.3 提交流程

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

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

4.3.2 yarn-cluster 提交流程 per-job

0. Flink 任务提交后，Client 向 HDFS 上传 Flink 的 Jar 包和配置
1. 向 Yarn RM 提交任务，RM 分配 Container 资源
2. 通知对应的 NM 启动 AM，AM 启动后加载 Flink 的 Jar 包和配置构建环境，然后启动JobManager（Dispatcher）
   1. Dispatcher启动JobMaster
3. JobMaster 向 Flink RM 申请资源
4. Flink RM 向 Yarn RM 申请资源启动 TaskManager
5. RM 分配 Container 资源后，由 AppMaster 通知资源所在节点的 NM 启动 TaskManager
6. TaskManager 注册 Slot
7. 发出提供 Slot 命令
8. TaskManager 向 JobMaster 提供 Slot
9. JobMaster 提交要在 Slot 中执行的任务