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 申请处理结果