深入理解 Flink（七）Flink Slot 管理详解

深入理解 Flink 系列文章已完结，总共八篇文章，直达链接：
深入理解 Flink （一）Flink 架构设计原理
深入理解 Flink （二）Flink StateBackend 和 Checkpoint 容错深入分析
深入理解 Flink （三）Flink 内核基础设施源码级原理详解
深入理解 Flink （四）Flink Time+WaterMark+Window 深入分析
深入理解 Flink （五）Flink Standalone 集群启动源码剖析
深入理解 Flink （六）Flink Job 提交和 Flink Graph 详解
深入理解 Flink （七）Flink Slot 管理详解
深入理解 Flink （八）Flink Task 部署初始化和启动详解

1.JobMaster 注册成功之后开始调度

JobMaster 中封装了一个 DefaultScheduler，在 DefaultScheduler.startSchedulingInternal() 方法中生成 ExecutionGraph 以执行调度。

2.Flink 的资源管理机制

资源调度的大体流程如下：

a.TaskExecutor 注册

• Register：当 TaskExecutor 启动之后，会向 ResourceManager 注册自己（TaskExecutor）和自己内部的 Slot（TaskManagerSlot）。
• Status Report：TaskExecutor 启动之后，会定期向 ResourceManager 进行心跳汇报，在心跳 payload 中，会携带 Slot 信息，ResourceManager 会随即更新自己内部 Slot 状态。

b.JobMaster/ JobManager 内部分配

• allocateSingleSlot：Scheduler 向 SlotPool 发送请求，如果 Slot 资源足够则直接分配，如果 Slot 资源不够，则由 SlotPool 再向 ResourceManager 中的 SlotManager 发送请求，此时即为 Job 向 Cluster 请求资源。
• allocateSharedSlot : Scheduler 向 SlotSharingManager 发送请求，SlotSharingManager 构建好 Slot 树之后， 向 SlotPool 发送请求，如果 Slot 资源足够则直接分配，如果 Slot 资源不够，则由 SlotPool（JobMaster） 再向 SlotManager（ResourceManager）发送请求，此时即为 Job 向 Cluster 请求资源。

c.Job 向 Cluster 请求资源

• 当通过 On YARN 的 per-job 模式 或者 sessionCluster 模式运行，则会在 YARN 集群中，启动一个 Flink 的小集群（申请一个 Continer 用来启动 JobManager 主节点，申请一堆 Container 用来启动 TaskManager 从节点）如果所有从节点提供的总 Slot 资源不够，则可以尝试从 YARN 中申请 Container 再启动 TaskManager 从节点，所以就多提供了一些 Slot 资源。直到 YARN 中的所有资源耗尽。 - 申请一个 Continer 用来启动 JobManager（Flink 主节点）- Flink 主节点向 YARN 主节点申请一堆 Container 用来启动 TaskManager（Flink 从节点）
• 如果 SlotManager 判断集群当中有足够的资源可以满足需求，那么就会向有资源的 TaskManager 发送 Request 指令，SlotPool 再去满足 Scheduler 的资源请求。
• 在 ActiveResourceManager 资源部署模式下，当 ResourceManager 判定 Flink Cluster 中没有足够的资源去满足需求时，它会进一步去底层的资源调度系统（YARN，K8s）请求资源，由调度系统把新的 TaskManager 启动起来，并且 TaskManager 向 Resource Manager 注册，则完成了新 Slot 的补充。
• 这种模式下，资源一共分为三级：SlotPool（job 已申请到的 slot）、SlotManager（cluster 已申请到的）、YARN（集群中剩余的）。

3.Flink 的 Slot 管理相关 RPC 请求

Flink 的 slot 管理分为 ResourceManager、TaskExecutor、JobMaster 3 个部分。

a.ResourceManager

• ResourceManager 是资源管理者，通过 SlotManager 进行 slot 管理。
• 负责处理 TaskExecutor 的 Slot 注册和 Slot 汇报。
• 接收到 JobMaster 申请 slot 的请求，然后向 TaskExecutor 发送 RPC 请求申请 slot。

b.TaskExecutor

• TaskExecutor 是资源提供者，通过 SlotTable 进行 slot 管理。
• TaskExecutor 上线之初，会找 ResourceManager 进行 slot 注册，而后在心跳过程中，不停进行 SlotStatus 和 Payload 的 Report。
• TaskExecutor 接收到 ResourceManager 的分派 Slot 的 RPC 请求之后，完成 slot 分派，并且告知 JobMaster。

c.JobMaster

• JobMaster 是资源使用者，通过 SlotPool 进行 Slot 管理。
• JobMaster 负责向 ResourceManager 申请 slot，或者取消申请。
• JobMaster 会接收来自 TaskExecutor 的 slot 的分派完成，然后部署 Task 执行，当 Task 执行完毕之后，会释放 Slot。

关于 Flink 中 slot 的申请和释放的 RPC 调用关系：

1. TaskExecutor 上线之初，通过 sendSlotReport 向 ResourceManager 注册 Slot。
2. JobMaster 启动之后，通过 requestSlot 向 ResourceManager 申请 slot，申请期间，也可以通过 cancelSlotRequest 取消申请。
3. ResourceManager 接收到 JobMaster 的 slot 请求，会通过 requestSlot 将请求转发给 TaskExecutor。
4. TaskExecutor 接收到 ResourceManager 的 requestSlot 请求之后完成 slot 分派，然后通过 offerSlot 告知 JobMaster。
5. JobMaster 在使用完毕 slot 之后，会通过 freeSlot 向 TaskExecutor 释放 slot。
6. TaskExecutor 在接收到一个 freeSlot 请求之后，会通过 notifySlotAvailable 告知 ResourceManager 该 slot 可用。

JobMaster 申请 slot 的核心入口：

allocateSlots(executionVertexDeploymentOptions);

大体上分为四个大步骤：

1、JobMaster 发送请求申请 slot 的请求给 ResourceManager
2、ResourceManager 接收到请求，执行 slot 分派处理，然后转发请求给 TaskExecutor
3、TaskExecutor 处理 ResourceManager 发送过来的 Slot 请求，完成 Slot 分派
4、JobMaster 接收到 TaskExecutor 发送过来的 Slot 申请处理结果