1.3.3　Flink資源管理

Apache Flink是一個分布式系統，需要計算資源才能執行應用程序。Flink集成了所有常見的集群資源管理器，如Hadoop YARN、Apache Mesos和Kubernetes，不過也可以設置作為Standalone獨立集群運行。

在Flink中，資源是由TaskManager上的Slot來表示的，每個Slot可以用來執行不同的任務（Task），而Job中實際的Task包含了待執行的用戶邏輯代碼。作業調度的主要目的就是給Task找到匹配的Slot。實際上，Flink作業調度可以看作對資源和任務進行匹配的過程。

1．Task執行

這是很難解釋和理解的部分。在深入講解之前，讀者應該記住什么是Flink中的操作符和任務。一般情況下，需要記住：Operator＝（Task1＋Task2＋…＋TaskN）。

（1）Operator in Flink＝>將一個數據流轉換為另一個數據流（可以是相同類型的數據流，也可以是不同類型的數據流）。Operator（操作符）是邏輯數據流圖（也稱為JobGraph）的節點。

（2）Task in Flink＝>是由Flink的運行時執行的基本工作單元。任務是物理數據流圖（也稱為Execution Graph）的節點。

（3）任務是操作符或操作符鏈的一個并行實例（兩個或多個連續操作符之間沒有任何重分區）。

（4）在Apache Flink中，每個Task只有一個線程，Task之間沒有共享知識。例如，Task1不知道另一個Task正在發生什么。這意味著API可以從一個Task中訪問每種狀態段，但沒有辦法訪問其他線程中的狀態。

（5）還有子任務（Sub-Task）的概念。子任務是在數據流的一部分上工作的任務。子任務指的是同一個操作符或操作符鏈有多個并行任務（這實際上意味著存在數據并行性）。

了解了術語操作符（Operator）、任務（Task）和子任務（Sub-Task）之后，接下來了解任務執行部分，包括：

（1）TaskManager（worker/slave進程）可以同時執行多個任務，并發數通常與TaskManager所在機器的CPU數有關。例如，如果機器的CPU數是16，則一個TaskManager可以同時運行16個任務。這是Apache Flink提供的最佳解決方案，可以同時運行16個任務，但這可能會導致一些問題，如Flink集群可能會經常重新啟動自己等。

（2）一個TaskManager在同一個JVM進程中以多線程方式執行它的任務。

（3）任務可以是相同操作符（記住數據并行性）的子任務，也可以是不同操作符（記住任務并行性）的子任務，甚至可以是來自不同應用程序的子任務。

（4）TaskManager提供一定數量的槽位（與機器的CPU數量相關）來控制它能夠并發執行的任務。換句話說，如果機器有16個CPU，則TaskManager可以有16個槽，每個槽用于處理一個任務。

2．操作符鏈

對于分布式執行，Flink將操作子任務連成Tasks。每個Task由一個線程執行。將操作符連接到Tasks中是一種優化行為，它減少了線程到線程切換和緩沖的開銷，并在降低時延的同時增加了總體吞吐量。

例如，一個數據流使用5個子任務執行，因此使用5個并行線程，如圖1-10所示。

操作符鏈允許非shuffle操作在同一個線程中共存，完全避免了序列化和反序列化。

3．任務槽

每個TaskManager（Worker進程）都是一個JVM進程，可以在單獨的線程中執行一個或多個子任務。為了控制一個worker接受多少任務，一個worker具有至少一個“任務插槽”。

每個Task Slot表示TaskManager資源的一個固定子集。例如，一個TaskManager有3個插槽，它會將其1/3的托管內存分配給每個插槽。對資源進行插槽化意味著子任務不會與來自其他作業的子任務爭奪托管內存，而是擁有一定數量的預留托管內存。注意，這里沒有發生CPU隔離；當前插槽只分隔任務的托管內存，如圖1-11所示。

通過調整任務槽的數量，用戶可以定義子任務如何彼此隔離。每個TaskManager有一個插槽（Slot）意味著每個任務組運行在各自的JVM中（例如，可以在單獨的容器中啟動JVM）。擁有多個插槽意味著更多的子任務共享同一個JVM。相同JVM中的任務共享TCP連接（通過多路復用）和心跳消息。它們還可以共享數據集和數據結構，從而減少每個任務的開銷。

默認情況下，Flink允許子任務共享槽位，即使它們是不同任務的子任務，只要它們來自相同的作業。結果是一個槽位可以容納作業的整個管道，如圖1-12所示。

這種槽位共享有以下兩個主要好處：

（1）Flink集群需要的任務插槽與作業中使用的最高并行度一樣多。不需要計算一個程序總共包含多少任務（具有不同的并行度）。

（2）更容易得到更好的資源利用。如果沒有槽位共享，則非密集型source/map（）子任務將阻塞與資源密集型窗口子任務一樣多的資源。使用插槽共享，將示例中的基本并行度從2提高到6，可以充分利用插槽資源，同時確保繁重的子任務在TaskManager中得到公平分配。

API還包括一個資源組（Resource Group）機制，可用于防止不需要的槽位共享。

根據經驗，恰當的默認任務槽位數應該是CPU核的數量。使用超線程，每個槽位將接受2個或更多的硬件線程上下文。

4．資源申請

在ResourceManager中，有一個子組件叫作SlotManager，SlotManager用于維護當前集群中所有TaskManager上的Slot的信息與狀態，例如該Slot在哪個TaskManager中，該Slot當前是否空閑等，如圖1-13所示。

當JobMaster為特定Task申請資源時，根據當前作業部署模式（關于作業部署模式，可參閱1.5節）的區別，TaskManager可能已經啟動或者尚未啟動。如果TaskManager尚未啟動，則ResourceManager會去申請資源來啟動新的TaskManager。當TaskManager啟動之后，它會通過服務找到當前活躍的ResourceManager并進行注冊。在注冊信息中，會包含該TaskManager中所有Slot的信息。ResourceManager收到注冊信息后，其中的SlotManager就會記錄下相應的Slot信息。當JobMaster為某個Task來申請資源時，SlotManager就會從當前空閑的Slot中按一定規則選擇一個空閑的Slot進行分配。當分配完成后，ResourceManager會首先向TaskManager發送RPC要求將選定的Slot分配給特定的JobManager。TaskManager如果還沒有執行過該JobMaster的Task，則它首先需要與相應的JobMaster建立連接，然后發送提供Slot的RPC請求。在JobMaster中，所有Task的請求會緩存到SlotPool中。當有Slot被提供之后，SlotPool會從緩存的請求中選擇相應的請求并結束相應的請求過程。

當Task結束之后，無論是正常結束還是異常結束，都會通知JobMaster相應的結束狀態，然后在TaskManager端將Slot標記為已占用但未執行任務的狀態。JobMaster會首先將相應的Slot緩存到SlotPool中，但不會立即釋放。這種方式避免了如果將Slot直接還給ResourceManager，在任務異常結束之后需要重啟時，則需要立刻重新申請Slot的問題。通過延時釋放，容錯的Task可以盡快調度回原來的TaskManager，從而加快故障切換的速度。當SlotPool中緩存的Slot超過指定的時間仍未使用時，SlotPool就會發起釋放該Slot的過程。與申請Slot的過程對應，SlotPool會首先通知TaskManager來釋放該Slot，然后TaskManager通知ResourceManager該Slot已經被釋放，從而最終完成釋放的邏輯。

5．心跳報告

除了正常的通信邏輯外，在ResourceManager和TaskManager之間還存在定時的心跳消息同步Slot的狀態。在分布式系統中，消息的丟失、錯亂不可避免，這些問題會在分布式系統的組件中引入不一致狀態，如果沒有定時消息，則組件無法從這些不一致狀態中恢復。此外，當組件之間長時間未收到對方的心跳時，就會認為對應的組件已經失效，并進入容錯的流程。

6．共享槽位

在Slot管理基礎上，Flink可以將Task調度到相應的Slot當中。如上文所述，Flink尚未完全引入細粒度的資源匹配，默認情況下，每個Slot可以分配給一個Task，但是，這種方式在某些情況下會導致資源利用率不高，如圖1-14所示，假如A、B、C依次執行計算邏輯，那么給A、B、C分配單獨的Slot就會導致資源利用率不高。為了解決這一問題，Flink提供了共享槽位（Share Slot）的機制，如圖1-14所示。基于共享槽位，每個Slot中可以部署來自不同JobVertex（作業向量）的多個任務，但是不能部署來自同一個JobVertex的Task，如圖中所示，每個Slot中最多可以部署同一個A、B或C的Task，但是可以同時部署A、B和C的各一個Task。當單個Task占用資源較少時，共享槽位可以提高資源利用率。此外，共享槽位也提供了一種簡單的保持負載均衡的方式。