Flink 1.12 資源管理新特性回顧
本文由社區志愿者陳政羽整理，Apache Flink Committer、阿里巴巴技術專家宋辛童，Apache Flink Contributor、阿里巴巴高級開發工程師郭旸澤分享，主要介紹 Flink 1.12 資源管理的一些特性。內容主要分為 4 部分：
1.內存管理
2.資源調度擴展
3.資源框架
4.未來規劃
一、內存管理
首先回顧 Flink 的內存模型變遷。下圖左邊分別為 Flink 1.10、Flink 1.11 引入的新的內存模型。盡管涉及的模塊較多，但 80% - 90% 的用戶僅需關注真正用于任務執行的 Task Heap Memory、Task Off-Heap Memory、Network Memory、Managed Memory 四部分。
其它模塊大部分是 Flink 的框架內存，正常不需要調整，即使遇到問題也可以通過社區文檔來解決。除此之外，“一個作業究竟需要多少內存才能滿足實際生產需求” 也是大家不得不面臨的問題，比如其他指標的功能使用、作業是否因為內存不足影響了性能，是否存在資源浪費等。
針對上述內容，社區在 Flink 1.12 版本提供了一個全新的， 關于 Task manager 和 Job
manager 的 Web UI。
在新的 Web UI 中，可以直接將每一項監控指標配置值、實際使用情況對應到內存模型中進行直觀的展示。在此基礎上，可以更清楚的了解到作業的運行情況、該如何調整、用哪些配置參數調整等 (社區也有相應的文檔提供支持)。通過新的 Web UI，大家能更好的了解作業的使用情況，內存管理也更方便。
1. 本地內存（Managed Memory）
Flink 托管內存實際上是 Flink 特有的一種本地內存，不受 JVM 和 GC 的管理，而是由 Flink 自行進行管理。
本地內存的特點主要體現在兩方面：
一方面是 slot 級別的預算規劃，它可以保證作業運行過程中不會因為內存不足，造成某些算子或者任務無法運行；也不會因為預留了過多的內存沒有使用造成資源浪費。 同時 Flink 能保證當任務運行結束時準確將內存釋放，確保 Task Manager 執行新任務時有足夠的內存可用。
另一方面，資源適應性也是托管內存很重要的特性之一，指算子對于內存的需求是動態可調整的。具備了適應性，算子就不會因為給予任務過多的內存造成資源使用上的浪費，也不會因為提供的內存相對較少導致整個作業無法運行，使內存的運用保持在一定的合理范圍內。當然，在內存分配相對比較少情況下，作業會受到一定限制，例如需要通過頻繁的落盤保證作業的運行，這樣可能會影響性能。
當前，針對托管內存，Flink 的使用場景如下：
RocksDB 狀態后端：在流計算的場景中，每個 Slot 會使用 State 的 Operator，從而共享同一底層 的 RocksDB 緩存；
Flink 內置算子：包含批處理、Table SQL、DataSet API 等算子，每個算子有獨立的資源預算，不會相互共享；
Python 進程：用戶使用 PyFlink，使用 Python 語言定義 UDF 時需要啟動 Python 的虛擬機進程。
2. Job Graph 編譯階段
Flink 對于 management memory 的管理主要分為兩個階段。
2.1 作業的 Job Graph 編譯階段
在這個階段需要注意三個問題：
第一個問題是：slot 當中到底有哪些算子或者任務會同時執行。這個問題關系到在一個查詢作業中如何對內存進行規劃，是否還有其他的任務需要使用 management memory，從而把相應的內存留出來。 在流式的作業中，這個問題是比較簡單的，因為我們需要所有的算子同時執行，才能保證上游產出的數據能被下游及時的消費掉，這個數據才能夠在整個 job grep 當中流動起來。 但是如果我們是在批處理的一些場景當中，實際上我們會存在兩種數據 shuffle 的模式，一種是 pipeline 的模式，這種模式跟流式是一樣的，也就是我們前面說到的 bounded stream 處理方式，同樣需要上游和下游的算子同時運行，上游隨時產出，下游隨時消費。另外一種是所謂的 batch 的 blocking的方式，它要求上游把數據全部產出，并且落盤結束之后，下游才能開始讀數據。這兩種模式會影響到哪些任務可以同時執行。目前在 Flink 當中，根據作業拓撲圖中的一個邊的類型 (如圖上)。我們劃分出了定義的一個概念叫做 pipelined region，也就是全部都由 pipeline 的邊鎖連通起來的一個子圖，我們把這個子圖識別出來，用來判斷哪些 task 會同時執行。
第二個問題是：slot 當中到底有哪些使用場景？我們剛才介紹了三種 manage memory 的使用場景。在這個階段，對于流式作業，可能會出現 Python UDF 以及 Stateful Operator。這個階段當中我們需要注意的是，這里并不能肯定 State Operator 一定會用到 management memory，因為這跟它的狀態類型是相關的。如果它使用了 RocksDB State Operator，是需要使用 manage memory 的；但是如果它使用的是 Heap State Backend，則并不需要。然而，作業在編譯的階段，其實并不知道狀態的類型，這里是需要去注意的地方。
第三個問題：對于 batch 的作業，我們除了需要清楚有哪些使用場景，還需要清楚一件事情，就是前面提到過 batch 的 operator。它使用 management memory 是以一種算子獨享的方式，而不是以 slot 為單位去進行共享。我們需要知道不同的算子應該分別分配多少內存，這個事情目前是由 Flink 的計劃作業來自動進行設置的。
2.2 執行階段
第一個步驟是根據 State Backend 的類型去判斷是否有 RocksDB。如上圖所示，比如一個 slot，有 ABC 三個算子，B 跟 C 都用到了 Python，C 還用到了 Stateful 的 Operator。這種情況下，如果是在 heap 的情況下，我們走上面的分支，整個 slot 當中只有一種在使用，就是Python。之后會存在兩種使用方式：
其中一個是 RocksDB State Backend，有了第一步的判斷之后，第二步我們會根據用戶的配置，去決定不同使用方式之間怎么樣去共享 slot 的 management memory。
在這個 Steaming 的例子當中，我們定義的 Python 的權重是 30%，State Backend 的權重是 70%。在這樣的情況下，如果只有 Python，Python 的部分自然是使用 100% 的內存（Streaming 的 Heap State Backend 分支）；
而對于第二種情況（Streaming 的 RocksDB State Backend 分支），B、C 的這兩個 Operator 共用 30% 的內存用于 Python 的 UDF，另外 C 再獨享 70% 的內存用于 RocksDB State Backend。最后 Flink 會根據 Task manager 的資源配置，一個 slot 當中有多少 manager memory 來決定每個 operator 實際可以用的內存的數量。
批處理的情況跟流的情況有兩個不同的地方，首先它不需要去判斷 State Backend 的類型，這是一個簡化； 其次對于 batch 的算子，上文提到每一個算子有自己獨享的資源的預算，這種情況下我們會去根據使用率算出不同的使用場景需要多少的 Shared 之后，還要把比例進一步的細分到每個 Operator。
3. 參數配置
上方圖表展示了我們需要的是 manager，memory 大小有兩種配置方式：
一種是絕對值的配置方式，
還有一種是作為 Task Manager 總內存的一個相對值的配置方式。
taskmanager.memory.managed.consumer-weight 是一個新加的配置項，它的數據類型是 map 的類型，也就是說我們在這里面實際上是給了一個 key 冒號 value，然后逗號再加上下一組 key 冒號 value 的這樣的一個數據的結構。這里面我們目前支持兩種 consumer 的 key：
一個是 DATAPROC， DATAPROC 既包含了流處理當中的狀態后端 State Backend 的內存，也包含了批處理當中的 Batch Operator；
另外一種是 Python。
二、 資源調度
部分資源調度相關的 Feature 是其他版本或者郵件列表里面大家詢問較多的，這里我們也做對應的介紹。
1. 最大 Slot 數
Flink 在 1.12 支持了最大 slot 數的一個限制（slotmanager.number-of-slots.max），在之前我們也有提到過對于流式作業我們要求所有的 operator 同時執行起來，才能夠保證數據的順暢的運行。在這種情況下，作業的并發度決定了我們的任務需要多少個 slot 和資源去執行作業。
然而對于批處理其實并不是這樣的，批處理作業往往可以有一個很大的并發度，但實際并不需要這么多的資源，批處理用很少的資源，跑完前面的任務騰出 Slot 給后續的任務使用。通過這種串行的方式去執行任務能避免 YARN/K8s 集群的資源過多的占用。目前這個參數支持在 yarn/mesos/native k8 使用。
2. TaskManager 容錯
在我們實際生產中有可能會有程序的錯誤、網絡的抖動、硬件的故障等問題造成 TaskManager 無法連接，甚至直接掛掉。我們在日志中常見的就是 TaskManagerLost 這樣的報錯。對于這種情況需要進行作業重啟，在重啟的過程中需要重新申請資源和重啟 TaskManager 進程，這種性能消耗代價是非常高昂的。
對于穩定性要求相對比較高的作業，Flink1.12 提供了一個新的 feature，能夠支持在 Flink 集群當中始終持有少量的冗余的 TaskManager，這些冗余的 TaskManager 可以用于在單點故障的時候快速的去恢復，而不需要等待一個重新的資源申請的過程。
通過配置 slotmanager.redundant-taskmanager-num 可以實現冗余 TaskManager。這里所謂的冗余 TaskManager 并不是完完全全有兩個 TaskManager 是空負載運行的，而是說相比于我所需要的總共的資源數量，會多出兩個 TaskManager。
任務可能是相對比較均勻的分布在上面，在能夠在利用空閑 TaskManager 的同時，也能夠達到一個相對比較好的負載。 一旦發生故障的時候，可以去先把任務快速的調度到現有的還存活的 TaskManager 當中，然后再去進行新一輪的資源申請。目前這個參數支持在 yarn/mesos/native k8 使用。
3. 任務平鋪分布
任務平鋪問題主要出現在 Flink Standalone 模式下或者是比較舊版本的 k8s 模式部署下的。在這種模式下因為事先定義好了有多少個 TaskManager，每個 TaskManager 上有多少 slot，這樣會導致經常出現調度不均的問題，可能部分 manager 放的任務很滿，有的則放的比較松散。
在 1.11 的版本當中引入了參數 cluster.evenly-spread-out-slots，這樣的參數能夠控制它，去進行一個相對比較均衡的調度。
注意：
第一，這個參數我們只針對 Standalone 模式，因為在 yarn 跟 k8s 的模式下，實際上是根據你作業的需求來決定起多少 task manager 的，所以是先有了需求再有 TaskManager，而不是先有 task manager，再有 slot 的調度需求。在每次調度任務的時候，實際上只能看到當前注冊上來的那一個 TaskManager，Flink 沒辦法全局的知道后面還有多少 TaskManager 會注冊上來，這也是很多人在問的一個問題，就是為什么特性打開了之后好像并沒有起到一個很好的效果，這是第一件事情。
第二個需要注意的點是，這里面我們只能決定每一個 TaskManager 上有多少空閑 slot，然而并不能夠決定每個 operator 有不同的并發數，Flink 并不能決定說每個 operator 是否在 TaskManager 上是一個均勻的分布，因為在 flink 的資源調度邏輯當中，在整個 slot 的 allocation 這一層是完全看不到 task 的。
三、擴展資源框架
1. 背景
近年來，隨著人工智能領域的不斷發展，深度學習模型已經被應用到了各種各樣的生產需求中，比較典型的場景如推薦系統，廣告推送，智能風險控制。這些也是 Flink 一直以來被廣泛使用的場景，因此，支持人工智能一直以來都是 Flink 社區的長遠目標之一。針對這個目標，目前已經有了很多第三方的開源擴展工作。由阿里巴巴開源的工作主要有兩個：
一個是 Flink AI Extended 的項目，是基于 Flink 的深度學習擴展框架，目前支持 TensorFlow、PyTorch 等框架的集成，它使用戶可以將 TensorFlow 當做一個算子，放在 Flink 任務中。
另一個是 Alink，它是一個基于 Flink 的通用算法平臺，里面也內置了很多常用的機器學習算法。
以上的兩個工作都是從功能性上對 Flink 進行擴展，然而從算力的角度上講，深度學習模型亦或機器學習算法，通常都是整個任務的計算瓶頸所在。GPU 則是這個領域被廣泛使用用來加速訓練或者預測的資源。因此，支持 GPU 資源來加速計算是 Flink 在 AI 領域的發展過程中必不可少的功能。
2. 使用擴展資源
目前 Flink 支持用戶配置的資源維度只有 CPU 與內存，而在實際使用中，不僅是 GPU，我們還會遇到其他資源需求，如 SSD 或 RDMA 等網絡加速設備。因此，我們希望提供一個通用的擴展資源框架，任何擴展資源都可以以插件的形式來加入這個框架，GPU 只是其中的一種擴展資源。
對于擴展資源的使用，可以抽象出兩個通用需求：
需要支持該類擴展資源的配置與調度。用戶可以在配置中指明對這類擴展資源的需求，如每個 TaskManager 上需要有一塊 GPU 卡，并且當 Flink 被部署在 Kubernetes/Yarn 這類資源底座上時，需要將用戶對擴展資源的需求進行轉發，以保證申請到的 Container/Pod 中存在對應的擴展資源。
需要向算子提供運行時的擴展資源信息。用戶在自定義算子中可能需要一些運行時的信息才能使用擴展資源，以 GPU 為例，算子需要知道它內部的模型可以部署在那一塊 GPU 卡上，因此，需要向算子提供這些信息。
3. 擴展資源框架使用方法
使用資源框架我們可以分為以下這 3 個步驟：
首先為該擴展資源設置相關配置；
然后為所需的擴展資源準備擴展資源框架中的插件；
最后在算子中，從 RuntimeContext 來獲取擴展資源的信息并使用這些資源
3.1 配置參數
- # 定義擴展資源名稱，“gpu”external-resources: gpu# 定義每個 TaskManager 所需的 GPU 數量external-resource.gpu.amount: 1 # 定義Yarn或Kubernetes中擴展資源的配置鍵external-resource.gpu.yarn.config-key: yarn.io/gpuexternal-resource.gpu.kubernetes.config-key: nvidia.com/gpu# 定義插件 GPUDriver 的工廠類。external-resource.gpu.driver-factory.class: org.apache.flink.externalresource.gpu.GPUDriverFactory
以上是使用 GPU 資源的配置示例：
對于任何擴展資源，用戶首先需要將它的名稱加入 "external-resources" 中，這個名稱也會被用作該擴展資源其他相關配置的前綴來使用。示例中，我們定義了一種名為 "gpu" 的資源。
在調度層，目前支持用戶在 TaskManager 的粒度來配置擴展資源需求。示例中，我們定義每個 TaskManager 上的 GPU 設備數為 1。
將 Flink 部署在 Kubernetes 或是 Yarn 上時，我們需要配置擴展資源在對應的資源底座上的配置鍵，以便 Flink 對資源需求進行轉發。示例中展示了 GPU 對應的配置。
如果提供了插件，則需要將插件的工廠類名放入配置中。
3.2 前置準備
在實際使用擴展資源前，還需要做一些前置準備工作，以 GPU 為例：
在 Standalone 模式下，集群管理員需要保證 GPU 資源對 TaskManager 進程可見。
在 Kubernetes 模式下，需要集群支持 Device Plugin[6]，對應的 Kubernetes 版本為 1.10，并且在集群中安裝了 GPU 對應的插件。
在 Yarn 模式下，GPU 調度需要集群 Hadoop 版本在 2.10 或 3.1 以上，并正確配置了 resource-types.xml 等文件。
3.3 擴展資源框架插件
完成了對擴展資源的調度后，用戶自定義算子可能還需要運行時擴展資源的信息才能使用它。擴展資源框架中的插件負責完成該信息的獲取，它的接口如下：
- public interface ExternalResourceDriverFactory { /** * 根據提供的設置創建擴展資源的Driver */ ExternalResourceDriver createExternalResourceDriver(Configuration config) throws Exception;}public interface ExternalResourceDriver { /** * 獲取所需數量的擴展資源信息 */ Set<? extends ExternalResourceInfo> retrieveResourceInfo(long amount) throws Exception;}
ExternalResourceDriver 會在各個 TaskManager 上啟動，擴展資源框架會調用各個 Driver 的 retrieveResourceInfo 接口來獲得 TaskManager 上的擴展資源信息，并將得到的信息傳到算子的 RuntimeContext。ExternalResourceDriverFactory 則為插件的工廠類。
4. GPU 插件
Flink 目前內置了針對 GPU 資源的插件，其內部通過執行名為 Discovery Script 的腳本來獲取當前環境可用的 GPU 信息，目前信息中包含了 GPU 設備的 Index。
Flink 提供了一個默認腳本，位于項目的 "plugins/external-resource-gpu/" 目錄，用戶也可以實現自定義的 Discovery Script 并通過配置來指定使用自定義腳本。該腳本與 GPU 插件的協議為：
當調用腳本時，所需要的 GPU 數量將作為第一個參數輸入，之后為用戶自定義參數列表。
若腳本執行正常，則輸出 GPU Index 列表，以逗號分隔。
若腳本出錯或執行結果不符合預期，則腳本以非零值退出，這會導致 TaskManager 初始化失敗，并在日志中打印腳本的錯誤信息。
Flink 提供的默認腳本是通過 "nvidia-smi" 工具來獲取當前的機器中可用的 GPU 數量以及 index，并根據所需要的 GPU 數量返回對應數量的 GPU Index 列表。當無法獲取到所需數量的 GPU 時，腳本將以非零值退出。
GPU 設備的資源分為兩個維度，流處理器與顯存，其顯存資源只支持獨占使用。因此，當多個 TaskManager 運行在同一臺機器上時，若一塊 GPU 被多個進程使用，可能導致其顯存 OOM。因此，Standalone 模式下，需要 TaskManager 級別的資源隔離機制。
默認腳本提供了 Coordination Mode 來支持單機中多個 TaskManager 進程之間的 GPU 資源隔離。該模式通過使用文件鎖來實現多進程間 GPU 使用信息同步，協調同一臺機器上多個 TaskManager 進程對 GPU 資源的使用。
5. 在算子中獲取擴展資源信息
在用戶自定義算子中，可使用在 "external-resources" 中定義的資源名稱來調用 RuntimeContext 的 getExternalResourceInfos 接口獲取對應擴展資源的信息。以 GPU 為例，得到的每個 ExternalResourceInfo 代表一塊 GPU 卡，而其中包含名為 "index" 的字段代表該 GPU 卡的設備 Index。
- public class ExternalResourceMapFunction extends RichMapFunction<String, String> { private static finalRESOURCE_NAME="gpu"; @Override public String map(String value) { Set<ExternalResourceInfo> gpuInfos = getRuntimeContext().getExternalResourceInfos(RESOURCE_NAME); List<String> indexes = gpuInfos.stream() .map(gpuInfo -> gpuInfo.getProperty("index").get()).collect(Collectors.toList()); // Map function with GPU// ... }}
6. MNIST Demo
下圖以 MNIST 數據集的識別任務來演示使用 GPU 加速 Flink 作業。
MNIST 如上圖所示，為手寫數字圖片數據集，每個圖片可表示為為 28*28 的矩陣。在該任務中，我們使用預訓練好的 DNN 模型，圖片輸入經過一層全連接網絡得到一個 10 維向量，該向量最大元素的下標即為識別結果。
我們在一臺擁有兩塊 GPU 卡的 ECS 上啟動一個有兩個 TaskManager 進程的 Standalone 集群。借助默認腳本提供的 Coordination Mode 功能，我們可以保證每個 TaskManager 各使用其中一塊 GPU 卡。
該任務的核心算子為圖像識別函數 MNISTClassifier，核心實現如下所示
- class MNISTClassifier extends RichMapFunction<List<Float>, Integer> { @Override public void open(Configuration parameters) { //獲取GPU信息并且選擇第一塊GPU Set<ExternalResourceInfo> externalResourceInfos = getRuntimeContext().getExternalResourceInfos(resourceName); final Optional<String> firstIndexOptional = externalResourceInfos.iterator().next().getProperty("index"); // 使用第一塊GPU的index初始化JCUDA組件 JCuda.cudaSetDevice(Integer.parseInt(firstIndexOptional.get())); JCublas.cublasInit(); }}
在 Open 方法中，從 RuntimeContext 獲取當前 TaskManager 可用的 GPU，并選擇第一塊來初始化 JCuda 以及 JCublas 庫。
- class MNISTClassifier extends RichMapFunction<List<Float>, Integer> { @Override public Integer map(List<Float> value) { // 使用Jucblas做矩陣算法 JCublas.cublasSgemv('n', DIMENSIONS.f1, DIMENSIONS.f0, 1.0f, matrixPointer, DIMENSIONS.f1, inputPointer, 1, 0.0f, outputPointer, 1); // 獲得乘法結果并得出該圖所表示的數字 JCublas.cublasGetVector(DIMENSIONS.f1, Sizeof.FLOAT, outputPointer, 1, Pointer.to(output), 1); JCublas.cublasFree(inputPointer); JCublas.cublasFree(outputPointer); int result = 0; for (int i = 0; i < DIMENSIONS.f1; ++i) { result = output[i] > output[result] ? i : result; } return result; }}
在 Map 方法中，將預先訓練好的模型參數與輸入矩陣放入 GPU 顯存，使用 JCublas 進行 GPU 中的矩陣乘法運算，最后將結果向量從 GPU 顯存中取出并得到識別結果數字。
具體案例演示流程可以前往觀看視頻或者參考 Github 上面的鏈接動手嘗試。
四、未來計劃
除了上文介紹的這些已經發布的特性外，Apache Flink 社區也正在積極準備更多資源管理方面的優化特性，在未來的版本中將陸續和大家見面。
被動資源調度模式：托管內存使得 Flink 任務可以靈活地適配不同的 TaskManager/Slot 資源，充分利用可用資源，為計算任務提供給定資源限制下的最佳算力。但用戶仍需指定計算任務的并行度，Flink 需要申請到滿足該并行度數量的 TaskManager/Slot 才能順利執行。被動資源調度將使 Flink 能夠根據可用資源動態改變并行度，在資源不足時能夠 best effort 進行數據處理，同時在資源充足時恢復到指定的并行度保障處理性能。
細粒度資源管理：Flink 目前基于 Slot 的資源管理與調度機制，認為所有的 Slot 都具有相同的規格。對于一些復雜的規模化生產任務，往往需要將計算任務拆分成多個子圖，每個子圖單獨使用一個 Slot 執行。當子圖間的資源需求差異較大時，使用相同規格的 Slot 往往難以滿足資源效率方面的需求，特別是對于 GPU 這類成本較高的擴展資源。細粒度資源管理允許用戶為作業的子圖指定資源需求，Flink 會根據資源需求使用不同規格的 TaskManager/Slot 執行計算任務，從而優化資源效率。
五、總結
通過文章的介紹，相信大家對 Flink 內存管理有了更加清晰的認知。
首先從本地內存、Job Graph 編譯階段、執行階段來解答每個流程的內存管理以及內存分配細節，通過新的參數配置控制 TaskManager的內存分配；
然后從大家平時遇到資源調度相關問題，包括最大 Slot 數使用，如何進行 TaskManager 進行容錯，任務如何通過任務平鋪均攤任務資源；
最后在機器學習和深度學習領域常常用到 GPU 進行加速計算，通過解釋 Flink 在 1.12 版本如何使用擴展資源框架和演示 Demo， 給我們展示了資源擴展的使用。再針對資源利用率方面提出 2 個社區未來正在做的計劃，包括被動資源模式和細粒度的資源管理。
