基于異構Flink 集群的節點優先級調度策略

汪文豪，史雪榮

（1.南京工業大學 計算機科學與技術學院，南京 211816；2.鹽城師范學院 數學與統計學院，江蘇 鹽城 224002）

0 概述

近年來，云計算、大數據、物聯網、人工智能等新一代信息技術快速發展，信息產業的數據量急劇增長。截止到2020 年底，我國的數據總量預計到達8 060 EB［1］。隨著信息數據量劇增，數據處理技術也在發生著巨大變化，涌現出Apache Hadoop、Apache Spark、Apache Storm、Heron、Flink 等［2］一大批數據計算系統。在數據處理的硬件方面，各種機構和設備的更新換代和新興技術的引入，使得各個數據計算系統在實際生產環境中變得越來越異構［3-4］。這種異構集群特征直觀表現在其節點的CPU、內存等各方面存在差異致使集群運行時的處理能力不盡相同，從而使得集群在調度節點執行任務時，集群效率明顯下降［5］。

目前，國內外眾多學者對資源彈性管理問題和分布式框架下的任務及節點調度問題展開研究。在資源彈性管理方面：文獻［6］提出使用排隊理論進行建模，通過部署主動彈性控制器和反應彈性控制器相結合的模型來估計集群節點的未來負載；文獻［7］認為云計算存在資源傾斜的問題，提出動態實時云框架，使集群可以自動適應不同的工作負載，并根據需求重新分配資源；文獻［8］基于云計算通用工作負載預測器，設計一種可以為不同工作負載提供高精度預測的長期記憶模型，解決了通用預測器精度不夠的問題，進一步優化了集群資源彈性管理。在批處理框架方面：文獻［9］認為Hadoop 框架在默認調度時未考慮Map 與Reduce 之間存在差異性，據此提出一種基于Hadoop 框架的截止時間約束的擴展MapReduce 任務調度算法；文獻［10］認為在Hadoop默認調度方式下缺乏對整體集群異構性的考慮，提出基于資源感知與資源異構的云計算環境任務調度算法；文獻［11］基于Hadoop 框架的節點計算能力，設計能夠按計算能力分配數據塊的數據局部性調度器；文獻［12］認為在異構Hadoop 集群中會出現節點隨任務動態分配而產生性能差異的問題，提出基于動態工作負載調整的任務調度策略；文獻［13］在YARN 資源調度器的基礎上，結合閉環反饋控制方法，使Hadoop 集群在運行時可以動態對MapReduce作業數進行控制，避免出現主觀性的問題；文獻［14］基于異構Spark 集群，提出通過監測節點資源的自適應動態節點任務調度策略，但是該策略依賴于人工設定的閾值，無法體現客觀性；文獻［15］在文獻［14］的基礎上，提出基于異構節點優先級的Spark 動態自適應調度算法，該算法能夠根據實時節點優先級完成調度，但是忽略了任務種類和集群整體工作環境，沒有徹底解決主觀性問題。在流處理框架方面：文獻［16］提出在流處理框架Storm on YARN 上構建一種雙層調度模型，通過對流數據處理的實時監測，做出合理的資源分配預測，解決框架需要人工干預調整的問題；文獻［17］針對流處理框架的數據拓撲中任務間通信開銷較大的問題，提出Flink 框架下的拓撲關鍵路徑模型，該模型能夠確保關鍵路徑上節點負載差異較小的同時最小化任務的節點間通信開銷。

但是現階段國內外的相關研究主要存在兩方面的問題。一方面，大部分的分布式調度研究主要關注于批處理，將Hadoop 框架與Spark 框架作為主要的實驗環境，而對于流處理，尤其是以Flink 框架作為實驗環境的研究相對較少。另一方面，在實際作業環境中，很難避免整體集群出現異構的情況，針對結合實際作業環境的異構Flink 環境的研究也相對較少。本文通過研究分布式節點的實時性能和集群的工作環境預測機制，充分考慮現實生產環境的異構分布問題，從而對Flink 底層默認調度策略進行優化，以提高Flink 系統的工作效率。

1 相關技術

Apache Flink 是一個面向數據流的有狀態計算框架，核心是一個流數據的處理引擎。在計算過程中，Flink 將所有任務當作流來處理，批處理任務被看作具備有限邊界的特殊數據流。相對于目前的大部分流處理框架，Flink 框架在數據處理方面有低延遲、高吞吐的優勢，在集群功能方面則提供了消息傳遞、狀態管理、容錯恢復等一系列服務［18-19］。

Flink 集群主要由兩個重要進程JobManager 和TaskManager 組成，如圖1 所示，其中虛線表示任務的流轉。JobManager 又稱為Master，主要協調整體架構的數據流圖的運算和執行，其中的調度器模塊和Checkpoint 協調器模塊還負責調度任務、協調檢查和故障恢復。TaskManager又稱為Worker，主要執行由JobManager分配的任務。同時，每一個TaskManager都具有緩存數據和節點間通信的功能。在Flink 中，一個TaskManager 即為一個JVM 進程，JVM 進程允許并行地執行子任務，并能夠指定若干slot。每一個slot 可以執行若干子任務，即運行若干線程［18］。

圖1 Flink 集群架構Fig.1 Flink cluster architecture

Flink 在調度任務分配slot 時，遵循2 個原則：1）同一Job 中的同一分組的不同task 可以共享同一slot；2）任務調度按照拓撲順序從Source 調度到Sink。以圖2（a）的Flink 拓撲模型為例，各頂點ν表示Flink 中的Operator 算子，νa表示Source 算子組件，νb、νc和νd表示并行度為3 的Transformation 算子組件，νe和νf表示并行度為2 的Sink 算子組件。首先由Source 組件將讀取的數據發送至Transformation 組件；然后Transformation 組件負責處理上游發送的數據；最后由Sink 組件接收上游處理結果并持久化至數據庫［20］。以有2 個TaskManager 的Flink 分布式集群為例，每一個TaskManager 配置有2 個slot。Flink在默認調度下對于該任務拓撲的slot 分配模型如圖2（b）所示。首先調度默認從拓撲的Source 任務開始，將νa隨機調度給任一slot，如圖2（b）中νa調度至slot11 所示；然后將νb、νc和νd調度至任意的slot，但νb、νc和νd同屬于一種Operator 算子，不能共享同一slot，因此將νb、νc和νd分別調度至slot11、slot12 和slot21；最后將νe和νf也分別調度至slot11 和slot12。

圖2 任務拓撲分配模型Fig.2 Task topology assignment model

2 基于節點優先級的Flink 動態調度策略

2.1 節點優先級定義及調整方法描述

2.1.1 相關定義

定義1（節點優先級）Flink 集群的節點優先級集合為P=｛P1，P2，…｝。Pix表示第i個節點的偏x相關性能優先級指數，當x=c時表示節點偏CPU 相關性能優先級指數，當x=m時表示節點偏內存相關性能優先級指數。

定義2（節點性能指數計算權值）在節點性能指數計算時，引入的各動靜態性能因素權值表示為向量與向量，當x=c時當前權值側重計算CPU相關優先性能指數，當x=m時當前權值側重計算內存相關優先性能指數。對于任意的，有；對于任意的。

定義3（節點資源因素指數［15］）動態性能指數D=｛D1，D2，…｝表示節點在執行任務時的實時性能變化。對于動態性能指數有3 個資源約束，分別是CPU 剩余率（CSR）、內存剩余率（MSR）和磁盤剩余率（DSR）。靜態性能指數S=｛S1，S2，…｝表示節點的不對稱性能因素。對于靜態性能指數有4 個資源約束，分別為CPU 速度（CS）、CPU 核數（CQ）、內存容量（MC）和磁盤容量（DC）。

對于每個動態指數Di∈D，對應資源約束向量di=(CCSRi，MMSRi，DDSRi)；對于每個靜態指數Si∈S，對應資源約束向量si=(CCSi，CCQi，MMCi，DDCi)。引入定義2中權值的計算后，得到Dix表示第i個節點的動態偏x性能指數，Six表示第i個節點的靜態偏x性能指數。

2.1.2 調整方法描述

本文所設計的節點優先級調整方法是即時反饋方法，節點優先級計算公式如下：

其中：α、β為節點整體動靜態因素指數權值，α+β=1；和表示節點動靜態偏x性能指數。和計算公式如下：

以4 個節點組成的Flink 異構集群為例，給出節點優先級調整架構如圖3 所示。

圖3 節點優先級調整架構Fig.3 Node priority adjustment architecture

監控器運行在集群Master 節點中，負責周期性收集節點的資源使用情況和任務隊列長度。獲取的資源信息將傳入到控制器中，由控制器完成具體節點優先級計算。監控器采用Ganglia［21］進行實現，Ganglia 是一個可擴展分布式集群資源監控系統，能夠實現對集群信息的監控。

控制器負責根據監控器監測到的信息計算出各節點的優先級。本文節點優先級調整方法引入作業環境預測機制，在一個周期內判定當前集群作業環境的參數x由式（4）確定。如果集群節點的I/O 操作平均時間T（I/O）超過了整體操作時間T（all）的65%，則判定當前集群作業環境為I/O 密集型，否則判定為CPU 密集型。

當監控器判定為I/O 密集型環境時，控制器將節點動靜態因素指數計算的權值重置為和，節點的偏內存性能優先級指數的計算公式如下：

當監控器判定為CPU 密集型環境時，控制器將節點動靜態因素指數計算的權值重置為和，節點的偏CPU 性能優先級指數的計算公式如下：

2.2 節點優先級調整算法

算法1節點優先級調整算法EP-NPAA

2.3 基于節點優先級的Flink 節點動態自適應調度策略

在實際運行環境中，異構分布式集群的節點往往會出現資源不平衡和負載不均衡的情況。為保證任務的高效完成，需要準確衡量各節點的性能以及整體集群的負載程度，選擇合適的節點分配任務［22］。本文提出一種基于節點優先級的Flink 節點動態自適應調度算法。

2.3.1 基于節點優先級的調度方法描述

基于節點優先級的Flink 節點動態自適應調度方法建立在集群異構的情況下，該方法的主要構成如下：1）Master 節點監控器定期監控每個Worker 節點自身資源情況及負載變化情況，并反饋控制器定期動態計算各Worker 節點的優先級指數；2）在Master 節點進行任務調度時，讀取各節點優先級指數，選擇指數較大的節點分配任務。Flink 節點動態自適應調度架構如圖4 所示。

圖4 Flink 節點動態自適應調度架構Fig.4 Dynamic adaptive Flink node scheduling architecture

2.3.2 Flink 節點動態自適應調度算法

Flink 節點動態自適應調度算法運行在Master節點上，具體步驟為：1）啟動集群，Master 節點檢測節點是否出現變化，根據變化狀態調用EP-NPAA 算法重新計算節點的靜態因素指數Si；2）Master 節點內控制器調用EP-NPAA 算法，依次計算每個Worker 節點的優先級得到P=｛P1，P2，…｝，并對節點集合NodeSets 進行排序；3）控制器從節點集合NodeSets中按優先級指數高低依次調用每個節點，檢測當前調用節點是否有當前task 可用的slot，若有可用的slot，則分配任務給該節點；若無可用slot，則繼續調用下一節點。

算法2Flink 節點動態自適應調度算法F-DASA

3 實驗與結果分析

3.1 實驗環境配置

為驗證本文自適應調度策略對Flink 異構集群的節點調度性能更佳，構建Flink 異構集群數據實驗平臺。該平臺由5 臺服務器組成，其中，1 臺為主服務器Master，4 臺為從服務器Worker。集群節點硬件與軟件配置如表1、表2 所示。

表1 集群節點硬件配置Table 1 Hardware configuration of cluster nodes

表2 集群節點軟件配置Table 2 Software configuration of cluster nodes

在實驗中進行節點優先級計算時，規定動靜態因素權值α和β分別取值為0.65 和0.35；在內存密集型作業環境下，各動態因素權值取值為（0.3，0.6，0.1），各靜態因素權值取值為（0.05，0.30，0.55，0.10）。在CPU 密集型作業環境下，各動態因素權值取值為（0.5，0.4，0.1），各靜態因素權值取值為（0.10，0.50，0.35，0.05）。

3.2 實驗數據設置

為有效驗證F-DASA 算法對Flink 異構集群的影響，實驗將本文策略與Flink 默認調度策略以及TSS-Flink 策略［17］進行大數據基準測試WorldCount和TeraSort 對比。WorldCount 是詞頻統計基準測試，特點是CPU 資源占用率較高、內存占用率較低，數據集采用9 個表生成模擬5 種事務處理的TPC-C 數據集［23］。TeraSort 是分布式排序基準測試，在執行過程中會大量占用內存資源，數據集為待排序的數值型數據集。

3.3 結果分析

3.3.1 運行時間對比

實驗通過對比使用Flink 默認調度策略、TSSFlink 策略和本文策略后的作業運行時間，驗證F-DASA 算法對Flink 異構集群的影響。在實驗過程中，基準測試WorldCount 的數據集規模分別為2 GB、4 GB 和6 GB，且設置不同的作業并行度，運行時間如圖5 所示。

圖5 作業運行時間對比Fig.5 Comparison of job runtime

從圖5 的實驗結果可以看出，在并行度為8 和16 的情況下，使用TSS-Flink 策略和本文策略后，基準測試WorldCount 的運行時間都有所減少。在使用本文策略之后，比使用Flink 默認調度策略約平均減少了6%。這是因為在默認調度策略下，調度器使異構集群中資源較少的節點完成和其他節點同等的任務，拖慢了整體作業運行時間。在本文策略下，調度器使任務得到均勻分配，資源多的節點可以完成盡可能多的任務，縮短了整體運行時間。相較于TSS-Flink 策略，本文策略的運行時間更少，這是因為TSS-Flink 策略缺少對于異構環境下節點資源不均衡問題的考慮，從而導致性能有所差異。

3.3.2 系統延遲對比

實驗通過對比Flink 默認調度策略與本文策略之間的延遲關系，驗證F-DASA 算法對Flink 異構集群的影響。在實驗過程中，在設置作業并行度為8的情況下，對比基準測試WorldCount 和TeraSort 下不同數據吞吐量的系統延遲，實驗結果如圖6所示。

圖6 2 種策略的系統延遲對比Fig.6 Comparison of system delay of two strategies

由圖6（a）可以看出，基準測試WorldCount 在使用默認調度策略時，隨著吞吐量的增加，系統延遲在100 ms 至150 ms 內緩慢上升。使用本文策略后，在吞吐量為1～6 萬組/s 時，系統延遲在100 ms 至150 ms 內緩慢上升，與Flink 默認調度策略的系統延遲相近；在吞吐量達到6 萬組/s 以上時，系統延遲上升幅度略微增大，屬于可接受范圍。由圖6（b）可以看出，基準測試TeraSort 在使用默認調度策略時，隨著吞吐量的增加，系統延遲始終在150 ms 上下浮動。在使用本文策略后，系統延遲在吞吐量為1～6 萬組/s時，在145 ms 上下浮動，略微優于Flink 默認調度策略；在吞吐量達到6 萬組/s 以上時，系統延遲上升幅度略微增大，屬于可接受范圍。綜上所述，在異構Flink 集群中使用本文策略后，依然能夠保持較為穩定的延遲率，達到集群原有的響應速度。

4 結束語

針對異構Flink 集群默認策略在節點調度過程中存在部分節點負載不均衡的問題，本文提出一種基于節點優先級的Flink 節點動態自適應調度策略。該策略能夠監控集群中任務與節點的各項數據，并在任務執行過程中根據實時的作業環境更新各個節點的優先級指數，為系統任務找到最佳的執行節點。實驗結果表明，該策略可在保持集群低延遲的基礎上，提高異構Flink 集群對于節點資源的利用率。下一步將針對節點的CPU、內存和帶寬性能設置合理的閾值，確保集群不會出現滿負載狀態，同時設計集群任務選擇算法，并將其與F-DASA 算法相結合進一步提升異構Flink 集群整體性能。