大家好，我是小寒~
在 Flink 中送出作業到 Flink 叢集後， Flink 叢集是如何為作業配置設定資源，以及如何管理叢集資源的呢？今天我們就來聊一聊 Flink 資源管理相關的内容。
資源管理
資源抽象
Flink 涉及的資源分為兩級：叢集資源和Flink自身資源。
叢集資源管理的是硬體資源，包括 CPU、記憶體、GPU等，由資源管理架構（Yarn、K8s、Mesos）來管理，Flink 從資源管理架構中申請和釋放資源。
Flink 從資源管理架構申請資源容器（Yarn 中的 Container），一個容器中運作一個TaskManager 程序。容器的資源對 Flink 來說也是比較粗粒度的。因為計算類型的不同，一個任務占用一個容器可能無法充分利用資源，是以單個容器會被多個Flink 的任務共享。
Flink 對申請到的資源進行切分，每一份叫作 TaskSlot。
從總體上來說，在資源管理中涉及了 JobMaster、ResourceManager、TaskManager 三種角色。JobMaster 是 Slot 資源的使用者，向 ResourceManager 申請資源，ResourceManager 負責配置設定資源和資源不足時申請資源，資源空閑時釋放資源。TaskManager 是 Slot 資源的持有者，在其 Slot 清單中記錄了 Slot 配置設定給了哪個作業的哪個 Task。
資料總管
資料總管在 Flink 中叫做 ResourceManager。Flink 同時支援不同的資源叢集類型。ResourceManager 位于 Flink 和 資源管理叢集（Yarn、K8s）之間，是 Flink 叢集級資源管理的抽象，其主要作用如下。
- 申請容器啟動新的TaskManager，或者為作業申請 Slot。
- 處理 JobManager 和 TaskManager 的異常退出。
- 緩存 TaskManager，等待一段時間之後再釋放掉不用的 TaskManager，避免資源反複地申請釋放。
- JobManager 和 TaskManager 的心跳感覺，對 JobManager 和 TaskManager 的退出進行對應的處理。
在 FlinK 中内置了 4 種 ResourceManager，即 YarnResourceManager、KubernetesResourceManager(K8s)、StandaloneResourceManager、 MesosResourceManager，分别對應于不同的資源管理架構。
Slot 管理器
Slot 管理器在 Flink 中叫作 SlotManager，是 ResourceManager 的元件，從全局角度維護目前有多少個 TaskManager，每個 TaskManager 有多少個空閑的 Slot 和 Slot 等資源的使用情況。當 Flink 作業排程執行時，根據 Slot 配置設定政策為 Task 配置設定執行位置。
SlotManager 雖然是 ResourceManager 的元件，但是其邏輯是通用的，并不關心到底使用了哪種資源叢集。面向不同的對象，SlotManager 提供不同的功能。
- 對 TaskManager 提供注冊、取消注冊、空閑退出等管理動作，注冊則叢集可用的 Slot 變多，取消注冊、空閑推出則釋放資源，還給資源管理叢集。
- 對 Flink 作業，接受 Slot 的請求和釋放、資源彙報等。當資源不足的時候，SlotManager 将資源請求暫存在等待隊列中，SlotManager 通知 ResourceManager 去申請更多的資源，啟動新的 TaskManager，TaskManager 注冊到 SlotManager 之後，SlotManager 就有可用的新資源了，從等待隊列依次配置設定資源。
Slot 選擇政策
Flink 在決定 Task 運作在哪個 TaskManager 上時，會根據政策進行選擇，選擇 Slot 的時候有不同的選擇政策。
選擇政策從總體上分為兩大類。
-
位置優先的選擇政策 LocationPreferenceSlotSelectionStrategy
位置優先的政策分為兩類:
- 預設政策（DefaultLocationPreferenceSlotSelectionStrategy），該政策不考慮資源的均衡配置設定，會從滿足條件的可用 Slot 集合選擇第一個，以此類雄。
- 均衡政策（EvenlySpreadOutLocationPreferenceSlotSelectionStrategy），該政策考慮資源的均衡配置設定，會從滿足條件的可用 Slot 集合中選擇剩餘資源最多的 Slot ，盡量讓各個 TaskManager 均衡地承擔計算壓力。
-
已配置設定 Slot 優先的選擇政策（PreviousAllocationSlotSelectionStrategy），如果目前沒有空閑的已配置設定的Slot，則仍然會使用位置
優先的政策來配置設定和申請 Slot。
Slot 資源池
Slot 資源池在 Flink 中叫作 SlotPool，是 JobMaster 中記錄目前作業從 TaskManager 擷取的 Slot 的集合。JobMaster 的排程器首先從SlotPool 中擷取 Slot 來排程任務，SlotPool 在沒有足夠的 Slot 資源執行作業的時候，首先會嘗試從 ResourceManager 中擷取資源。如果 ResourceManager 不可用、ResourceManager 拒絕資源請求 或 請求逾時，資源申請失敗，則作業啟動失敗。
JobMaster 申請到資源之後，會在本地持有 Slot，避免 ResourceManager 異常導緻作業運作失敗。對于批處理而言，持有資源 JobMaster 首先可以避免多次向 ResourceManager 申請資源，同時 ResourceManager 不可用也不會影響作業的繼續執行，隻有資源不足時才會導緻作業執行失敗。
當作業已經執行完畢或者作業完全啟動且資源有剩餘時，JobMaster 會将剩餘資源交還給 ResourceManager 。
Slot 共享
每一個 TaskManager 都是一個 java 程序，TaskManager 為每個 Task 配置設定一個線程。一個 TaskManager 中可能執行一個或者多個 Task。TaskManager 通過 Slot 來控制（一個 TaskManager 至少有一個 Slot）TaskManager 能夠接收多少個 Task。
Slot 表示 TaskManager 擁有資源的一個固定大小的子集。假如 一個 TaskManager 有 3 個 Slot，那麼它會将其管理的記憶體分成 3 份給各個 Slot，在沒有 Slot 共享的情況下，并行度為 2 的作業部署之後，Slot Task 的配置設定關系如下圖所示。
通過調整 Slot 的數量，使用者可以定義 Task 之間如何互相隔離。如果一個 TaskManager 隻有一個 Slot，意味着每個 Task 獨立地運作在 JVM 中。而一個 TaskManager 有多個 Slot，則意味着更多的 Task 可以共享一個 JVM。在同一個 JVM 程序中的 Task 将共享 TCP 連接配接和心跳消息。Task 之間也可能共享資料集和資料結構，這樣可以減少每個 Task 的負載。
雖然通過 Slot 對 TaskManager 的資源進行劃分，在一定程度上能夠提高叢集的計算資源使用率，但是這種做法并沒有考慮到不同 Task 的計算任務對資源需求的差異，計算任務有 IO 密集型、記憶體密集型、 CPU 密集型、GPU 密集型等不同的資源消耗類型，有時候還會是多種資源混合類型。
是以在 Slot 的基礎上，Flink 設計了 Slot 共享機制。其中，SlotSharingManager 用在 Flink 作業的執行排程中，負責 Slot 的共享，不同的 Task 可以共享 Slot。
Slot 共享的優點
預設情況下，Flink 作業共享同一個 SlotSharingGroup，同一個作業中來自不同 JobVertex 的 Task 可以共享 Slot。使用 Slot 共享，可以在一個 Slot 中運作 Task 組成的流水線。共享 Slot 帶來了如下優點。
-
資源配置設定簡單
Flink 叢集需要的 Slot 的數量和作業中的最高并行度一緻，不需要計算一個程式總共包含多少個 Task。
-
資源使用率高
如果沒有 Slot 共享，資源密集型的 Task （如長周期的視窗計算）跟非密集型的作業占用相同的資源，在整個 TaskManager 層面上，資源沒有充分利用。如果共享 Slot ，可以充分利用 Slot 資源。如下圖所示，将并行度從2提高到6，可以充分利用Slot資源，同時確定資源密集型的 Task 在TaskManager中公平配置設定。
Slot 共享組與 Slot 共享管理器
Slot 共享管理器在 Flink 中叫作 SlotSharingManager，Slot 共享組在 Flink 中叫做 SlotSharingGroup。SlotSharingManager 對象管理資源共享和配置設定，一個 Slot 共享組 對應一個 Slot 共享管理器。兩者在作業排程執行的時候發揮作用，部署 Task 之前，選擇 Slot 确定 Task 釋出到哪個TaskManager。
Flink 有兩種共享組。
-
SlotSharingGroup
非強制性共享限制，根據組内的 JobVertext ID 查找是否已有可以共享的 Slot，隻要確定相同 JobVertext ID 不能出現在一個共享的 Slot 内即可。
在符合資源要求的 Slot 中，找到沒有相同 JobVertext ID 的 Slot，根據 Slot 選擇政策選擇一個 Slot 即可，如果沒有符合條件的，則申請新的 Slot。
-
CoLocationGroup
CoLocationGroup 又叫作本地限制共享組，具有強制性的 Slot 共享限制。CoLocationGroup 用在疊代運算中，即在 IterativeStream 的 API 中調用。疊代運算中的 Task 必須共享同一個 TaskManager 的 Slot。CoLocationGroup 可以看成是 SlotSharingGroup 的特例。
此處需要注意， JobGraph 向 ExecutionGraph 的轉換過程中，為每一個 ExecutionVertext 賦予了按照并行度編寫的編号，相同編号的疊代計算 ExecutionVertext 會被放入本地共享限制組中，共享相同的 CoLocationConstraint 對象，在排程的時候，根據編号就能找到本組其他 Task 的 Slot 資訊。
CoLocation 共享根據組内每個 ExecutionVertext 關聯的 CoLocationConstraint 查找是否有相同 CoLocationConstraint 限制已配置設定 Slot 可用，在排程作業執行的時候，首先要找到本限制中其他 Task 部署的 TaskManager，如果沒有則申請一個新的 Slot，如果有則共享該 TaskManager 上的 Slot。
Slot 資源申請
-
單獨 Slot 資源申請
該類型的 Slot 申請首先會從 JobMaster 的目前 SlotPool 中嘗試擷取資源，如果資源不足，則 SlotPool 向 ResourceManager 請求新的 Slot。