导读：本文我们将重点介绍集群运行时中ResourceManager的设计和实现，了解如何通过ResourceManager对集群的计算资源进行有效管理。

作者：张利兵

来源：华章科技

01 ResourceManager详解

ResourceManager作为统一的集群资源管理器，用于管理整个集群的计算资源，包括CPU资源、内存资源等。

同时，ResourceManager负责向集群资源管理器中申请容器资源启动TaskManager实例，并对TaskManager进行集中管理。当新的作业提交到集群后，JobManager会向ResourceManager申请作业执行需要的计算资源，进而完成整个作业的运行。

如图3-12所示，为了兼容Hadoop Yarn、Kube.NETes、Mesos等集群资源管理器，在ResourceManager抽象实现类的基础上，分别实现了ActiveResourceManager、
Standalone-ResourceManager以及MesosResourceManager等子类。

其中ActiveResourceManager实现了动态资源管理，可以根据提交的作业动态选择启动或停止TaskManager实例。目前支持TaskManager动态管理和启动的ResourceManager主要有KubernetesResourceManager和Yarn-ResourceManager实现类。

▲图3-12 ResourceManager UML关系图

从图3-12中可以看出，ResourceManager通过实现ResourceManagerGateway接口，向其他组件提供RPC远程访问能力，如TaskManager服务和JobManager服务的Resource-ManagerGateway会将RPC访问请求发送到ResourceManager服务中。

另外，Resource-Manager继承了FencedRpcEndpoint基本实现类，使得ResourceManager可以作为一个RpcEndpoint节点，通过ResourceManagerGateway接口提供给其他服务节点，使之能够以RPC的方式访问ResourceManager服务。

同时，ResourceManager实现了LeaderContender接口，可以作为竞争节点让LeaderElectionService进行Leader节点的选举，保证整个集群ResourceManager组件服务的高可用。

从图3-12中也可以看出，ResourceManager主要包含如下成员变量。

• resourceId：ResourceManager对应的唯一资源ID。
• jobManagerRegistrations：专门存储JobManager注册信息。其中Key为JobID；Value为JobManagerRegistration，当启动JobManager服务时，就会将JobManager信息注册在jobManagerRegistrations实例中。
• jmResourceIdRegistrations：用于存储JobManager注册信息，与jobManagerRegistrations的区别在于Key为ResourceID。
• jobLeaderIdService：用于获取Job Leader ID的服务，在开启的高可用集群中，当JobManager的Leader节点发生切换时，会借助jobLeaderIdService获取当前作业有效的JobID和地址信息。
• taskExecutors：注册在ResourceManager的TaskExecutor列表中，其中Key为Task-Executor对应的ResourceID，Value为WorkRegistration，即TaskExecutor向Resource-Manager注册过程中所提供的信息。
• taskExecutorGatewayFutures：专门存储TaskExecutorGateway的CompletableFuture对象，Key为TaskExecutor对应的ResourceID，Value为CompletableFuture，用于获取Task-ExecutorGateway，实现与TaskExecutor之间的RPC通信。
• highAvailabilityServices：系统高可用服务，基于highAvailabilityServices服务支持组件高可用。
• heartbeatServices：用于创建HeartbeatManager服务，和其他组件之间建立心跳连接。
• fatalErrorHandler：系统异常错误处理，当ResourceManager出现异常时调用fatal-ErrorHandler处理异常错误。
• slotManager：ResourceManager的内部组件，用于管理集群的可用Slot资源，同时接收并处理TaskExecutor的SlotReport。
• clusterInformation：存储整个Flink集群共享的信息，包括blobServerHostname和blobServerPort等配置。
• resourceManagerMetricGroup：ResourceManager的MetricGroup，用于收集和Resource-Manager相关的监控指标。
• leaderElectionService：基于ZooKeeper实现的Leader选举服务，在这里用于实现Resource-Manager组件高可用。
• taskManagerHeartbeatManager：管理与TaskManager之间的心跳信息。
• jobManagerHeartbeatManager：管理与JobManager之间的心跳信息。
• clearStateFuture：用于停止ResourceManager后进行数据异步清理。

02 ResourceManagerGateway接口实现

ResourceManagerGateway接口提供了ResourceManager需要的RPC方法，供其他集群组件调用。例如在TaskExecutor中调用ResourceManagerGateway完成在ResourceManager中注册TaskExecutor的操作。

如图3-13所示，通过对ResourceManagerGateway中提供的RPC方法进行梳理，得到JobManager、TaskExecutor、WebMonitorEndpoint和Dispatcher等组件与ResourceManager-Gateway之间的RPC调用关系图。

▲图3-13 ResourceManager调用关系图

从图3-13中可以看出，JobManager、TaskExecutor、WebMonitorEndpoint和Dispatcher组件分别使用如下方法与ResourceManager服务进行交互。

1. JobManager和ResourceManager 的RPC调用

• registerJobManager()：在ResourceManager中注册JobManager服务，此时会在job-LeaderIdService服务中添加注册的JobManager信息。
• requestSlot()：JobManager向ResourceManager申请运行Task所需的Slot资源。
• heartbeatFromJobManager()：用于在JobManager与ResourceManager之间建立长期的心跳连接。
• disconnectJobManager()：根据JobID删除之前注册在ResourceManager中的Job-Manager信息，并且关闭JobManager与ResourceManager之间的RPC连接。

2. TaskExecutor和ResourceManager 的RPC调用

• heartbeatFromTaskManager()：在TaskExecutor中调用heartbeatFromTaskManager()方法，构建TaskExecutor与ResourceManager之间的心跳连接。
• disconnectTaskManager()：停止TaskExecutor组件时会调用disconnectTaskManager()方法断开TaskExecutor与ResourceManager之间的RPC连接。
• registerTaskExecutor()：当新的TaskExecutor启动时，会调用该方法向Resource-Manager注册TaskExecutor信息。
• sendSlotReport()：当TaskExecutor启动并注册成功后，会调用sendSlotReport()方法向ResourceManager上报SlotReport。SlotReport中包含TaskExecutor的资源数量和配置信息等内容。
• notifySlotAvailable()：当TaskExecutor中具有空闲Slot计算资源时，会调用notify-SlotAvailable()方法通知ResourceManager将该Slot资源变为Available状态。
• cancelSlotRequest()：取消JobManager已经分配的资源。

3. Dispatcher和ResourceManager的RPC调用

• requestResourceOverview()：用于在Dispatcher中获取集群资源信息，包括集群中的TaskManager、numberRegisteredSlots以及numberFreeSlots数量。
• requestTaskManagerMetricQueryServiceAddresses()：从ResourceManager获取Task-Manager的MetricQueryService路径，主要用于前端获取TaskManager的监控指标。

4. WebMonitorEndpoint和ResourceManager 的RPC调用

• requestTaskManagerInfo()：用于获取TaskManager的相关信息，即TaskExecutor启动过程中注册在ResourceManager的信息，包括TaskExecutor的网关地址、端口以及TaskExecutor的硬件信息。
• requestTaskManagerFileUpload()：请求上传文件到BlobServer上，返回Transient-BlobKey。

03 Slot计算资源管理

如图3-14所示，ResourceManager内部主要通过SlotManager服务统一对整个集群的Slot计算资源进行管理。Slot被称为资源卡槽，用于表示可以分配的最小计算资源单位，提交的Task最终会运行在Slot表示的计算资源中。

▲图3-14 Slot计算资源管理

从图3-14中可以看出，ResourceManager包含了Register Slot和Free Slot两个键值对集合。其中Register Slot专门存储ResourceManager中所有已经注册的TaskManagerSlot信息，Free Slot集合则存储了当前SlotManager中处于空闲状态且还没有被分配和使用的Slot集合。

TaskManagerSlot对象包含了SlotID、ResourceProfile以及TaskExecutorConnection等信息。如果Slot被分配使用，在TaskManagerSlot中还会存储AllocationID和JobID等分配信息，表明当前Slot已经被指定JobID对应的JobManager使用。

另外，SlotManager还包含了pendingSlotRequests和fulfilledSlotRequests两个键值对集合。其中pendingSlotRequests存储了所有处于pending和unfulfilled状态的Slot请求，fulfilledSlotRequests存储了所有已经分配完成的Slot请求。

Slot资源申请都会以Pending-SlotRequest的形式存储在pendingSlotRequests集合中，等待SlotManager根据当前集群的Slot资源进行分配。

当符合条件的Slot资源分配给指定的PendingSlotRequest后，会为其创建AllocationId，并将分配了AllocationId和SlotId信息的SlotRequest存储到fulfilled-SlotRequests集合中。

对Slot计算资源的注册和管理，主要是在TaskManager和ResourceManager服务之间进行的，TaskManager作为Slot计算资源的提供方，ResourceManager则作为Slot计算资源的接收和管理方。这里我们简单梳理一下TaskManager向SlotManager中注册Slot资源的整个过程。

• 启动TaskManager后，调用ResourceManagerGateway.registerTaskExecutor()方法向ResourceManager中注册TaskManager连接信息。
• 创建TaskManager和ResourceManager之间的RPC连接，TaskManager调用Resource-ManagerGateway.sendSlotReport()方法向ResourceManager发送SlotReport信息，接着ResourceManager调用SlotManager.registerTaskManager()方法，将TaskManager的资源信息写入SlotManager。
• 在SlotManager中根据SlotReport中的Slot信息创建TaskManagerSlot，并注册到SlotManager的HashMap<SlotID, TaskManagerSlot> slots集合中。
• SlotManager含有HashMap<SlotID, TaskManagerSlot> slots和LinkedHashMap<SlotID, TaskManagerSlot> freeSlots两个Slot集合。前者维护所有注册到SlotManager中的Slot计算资源，后者存储当前SlotManager中可用的Slot资源。

在SlotManager中完成Slot资源注册后，等待集群提交和运行作业。JobManager通过调用ResourceManagerGateway中的相关方法为作业申请Slot计算资源，整个申请过程如下。

• JobManager调用ResourceManagerGateway.requestSlot()方法向ResourceManager发起Slot计算资源申请。
• ResourceManager内部会调用SlotManager.registerSlotRequest()方法，向SlotManager申请作业需要的Slot计算资源。
• SlotManager中维护了HashMap<AllocationID, PendingSlotRequest> pendingSlotRequests集合，将所有的PendingSlotRequest存储在该集合中，并根据SlotRequest的Resource-Profile匹配合适的Slot计算资源，然后对Slot进行分配。
• 当SlotRequest需要的Slot计算资源分配完毕后，将已经分配的SlotID信息写入HashMap<AllocationID, SlotID> fulfilledSlotRequests集合。

SlotManager组件会对Slot进行统一的管理，在内部构建一个Slot计算资源池，有新的Slot注册时，会优先从pendingSlotRequests集合中获取处于Pending状态的SlotRequest，并为该SlotRequest分配Slot计算资源。

以上就是在ResourceManager中注册和分配Slot计算资源的全部过程，本文篇幅有限，Slot注册和分配过程中涉及的核心代码的详细介绍，请见《Flink设计与实现：核心原理与源码解析》第3.3.3节。

关于作者：张利兵，资深大数据专家和架构师，现任第四范式AI数据平台架构师，曾就职于明略数据。Apache Flink的贡献者，对Flink有非常深入的研究。长期从事大数据架构落地以及机器学习平台与数据平台研发架构工作，在Hadoop、Spark、机器学习等方面积累了丰富的经验。先后参与和主导了银行、证券、地铁等领域的大数据平台的架构设计与实现。《Flink原理、实战与性能优化》作者，极客时间《Flink原理与实战》专栏作者。

本书摘编自《Flink设计与实现：核心原理与源码解析》，经出版方授权发布。