Flink运行时调度过程图解

Posted 2021-11-24 monkeydbo

版本：Flink 1.13.2 - 2021-08-02

Flink 运行时集群的基本结构及调度过程图解

Flink 运行时集群的基本结构

针对不同集群环境（YARN，Mesos，Kubernetes，standalone等），结构会有略微不同，但是基本结构中包含了运行时的调度原理。

Flink Runtime 集群的基本结构，采用了标准 master-slave 的结构。中间 AM 中的部分表示 master，它负责管理整个集群中的资源和作业；而右侧的两个 TaskManager 则是 slave，负责提供具体的资源并实际执行作业任务。

途中 白色实线双线箭头表示数据交互，虚线表述调度过程。

AppMaster

Dispatcher

负责接收用户提供的作业，并且负责为这个新提交的作业拉起一个新的 JobManager 组件

ResourceManager

负责资源的管理，在整个 Flink 集群中只有一个 ResourceManager

JobManager

负责管理作业的执行，在一个 Flink 集群中可能有多个作业同时执行，每个作业都有自己的JobManager 组件

TaskManager

真正干活的组件，虽然他叫任务管理器，但是它并不管理任务，它是任务执行者，给他起名管理器大概是应为它并不是执行任务的最小单元，它里面还管理着插槽（Slot）。Slot 是 Flink 集群中最小的资源单元。

调度大致流程

基于上述结构，当用户提交作业的时候，提交脚本会首先启动一个 Client 进程负责作业的编译与提交。它首先将用户编写的代码编译为一个 JobGraph，在这个过程，它还会进行一些检查或优化等工作，例如判断哪些 Operator 可以 Chain 到同一个 Task 中。然后，Client 将产生的 JobGraph 提交到集群中执行。

此时有两种情况，一种是类似于 Standalone 这种 Session 模式，AM 会预先启动，此时 Client直接与 Dispatcher 建立连接并提交作业即可。另一种是 Per-Job 模式，AM 不会预先启动，此时 Client 将首先向资源管理系统 （如 Yarn、K8S）申请资源来启动AM，然后再向 AM 中的Dispatcher 提交作业。

资源管理与作业调度

在 Flink 中，资源是通过 Slot 来表示的，每个 Slot 可以用来执行不同的 Task。调度的主要目的就是为了给 Task 找到匹配的 Slot。

逻辑上来说，每个 Slot 都应对外描述能提供多少资源，而每个 Task 也应说明它需要申请到多少资源。但是实际上在 1.9之前，Flink 是不支持细粒度的资源描述的，而是统一的认为每个 Slot 提供的资源和Task 需要的资源都是相同的。从 1.9 开始，Flink 开始增加对细粒度的资源匹配的支持的实现，但这部分功能目前仍在完善中。

Flink 中资源管理的实现

图中：
ResourceManager 中，有一个子组件叫做 SlotManager，它维护了当前集群中所有 TaskExecutor 上的 Slot 的信息与状态，如该 Slot 在哪个TaskManager 中，该 Slot 当前是否空闲等。

JobManger 中，有一个 SlotPool 组件，它缓存了所有的 Slot 资源请求。

1. JobManger 为特定 Task 申请 Slot 资源，并且会将发起的请求缓存到 SlotPool 中

   此时会根据当前是 Per-job 还是 Session 模式，ResourceManager 可能会去申请资源来启动新的 TaskExecutor。（即后面的 2，3，4 步骤）

2. 向资源提供者（如K8s等）申请资源

3. 资源提供者已分配资源

4. 根据资源提供者分配的资源，启动 TM

5. 如果有 2，3，4 步骤（即启动新的 TM）则在启动 TM 后会向 RM 注册 Slot 信息。RM 根据现有的 Slot 信息向 TM 发起 Slot 请求

6. TM 向 JM 提供 Slot 资源，并且在 JM 获取到资源后会取消 SlotPool 中相应的 Slot 请求

7. 在 Task 结束后，无论是正常结束还是异常结束，都会通知 JobManager 相应的结束状态，然后在 TaskManager 端将 Slot 标记为已占用但未执行任务的状态。即：先将相应的 Slot 缓存到 SlotPool 中，但不会立即释放。通过延时释放，Failover 的 Task 可以尽快调度回原来的 TaskManager，从而加快 Failover 的速度。

8. 过一段时间后，如果延时释放的 Slot 没有在被使用，那么 TM 就会释放相应的 Slot 资源并向 RM 报告，跟新 Slot 的状态

9. 如果是通过资源提供者创建的新的 TM 那么还需要报告资源提供者，释放相应的 TM 资源。

另外：除了正常的通信逻辑外，在 ResourceManager 和 TaskManager 之间还存在定时的心跳消息来同步 Slot 的状态。

当 JobManger 来为特定 Task 申请资源的时候，根据当前是 Per-job 还是 Session 模式，ResourceManager 可能
会去申请资源来启动新的 TaskExecutor。

Task调度

在 Slot 管理基础上，Flink 可以将 Task 调度到相应的 Slot 当中。

虚线框表示 Task ，里面的圆形表示子任务，圆形里面的文字表示一些操作

从图中可以看出，有的任务会被合并到一个大任务中，即 Source 和 map 两个可以是独立的任务。

图中上面的是一个单一并行度的数量处理流程，每个 Task 都只需要一个 Slot 资源；下面的是一个带有并行度的数据处理流图，其中 【Source 、map】 和 【keyBy/window/apply】 两个任务的并行度都是 2 ，最后输出 sink 任务的并行度是 1。

从这里可以看出，slot 其实就是一个线程，把一些任务合并成一个大的任务放到统一个 Slot 中是很有优势的：可以降低线程切换带来的消耗，提高整体的吞吐量。至于什么样的 Task 会被合并到一个 Slot 中可以参考：chaining docs。

每个 TaskManager 都是一个 JVM 进程，它通过 Slot 数量来控制 TaskManager 能够接受多少任务。一个 Slot 并不是只能执行一个任务，只要重叠的任务，都可以在一个 Slot 中。在启动时，定义了 TM 的内存大小，这些内存会分到每一个 Slot 上，称之为委托内存。Slot 只在委托内存上进行相互的隔离，在 CPU 层面并不做隔离，所以，一般的建议都是：TM 的 Slot 数量和机器的 CPU 数量一致，这样可以降低计算资源的竞争。