導讀：本文我們將重點介紹集群運行時中ResourceManager的設計和實現，了解如何通過ResourceManager對集群的計算資源進行有效管理。

作者：張利兵

來源：華章科技

01 ResourceManager詳解

ResourceManager作為統一的集群資源管理器，用于管理整個集群的計算資源，包括CPU資源、內存資源等。

同時，ResourceManager負責向集群資源管理器中申請容器資源啟動TaskManager實例，并對TaskManager進行集中管理。當新的作業提交到集群后，JobManager會向ResourceManager申請作業執行需要的計算資源，進而完成整個作業的運行。

如圖3-12所示，為了兼容Hadoop Yarn、Kubernetes、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、機器學習等方面積累了豐富的經驗。先后參與和主導了銀行、證券、地鐵等領域的大數據平臺的架構設計與實現?！禙link原理、實戰與性能優化》作者，極客時間《Flink原理與實戰》專欄作者。

本書摘編自《Flink設計與實現：核心原理與源碼解析》，經出版方授權發布。