基于流網絡的Flink 平臺彈性資源調度策略

李梓楊，于炯,，卞琛，張譯天，蒲勇霖，王躍飛，魯亮

（1.新疆大學信息科學與工程學院，新疆 烏魯木齊 830046；2.新疆大學軟件學院，新疆 烏魯木齊 830008；3.廣東金融學院互聯網金融與信息工程學院，廣東 廣州 510521；4.中國民航大學計算機科學與技術學院，天津 300300）

1 引言

隨著云計算、物聯網、人工智能等新技術的不斷發展，自動駕駛、智慧城市、智能工業等新產業和新服務模式的不斷興起和發展，人們的生活方式也變得更舒適、更便捷。根據希捷公司發布的調查結果，預測到2025 年全球數據量將高達163 ZB，其中25%的數據需要被實時計算和處理，這些數據主要應用于物聯網[1]和人工智能等相關領域。

隨著數據規模和計算復雜度的提升，以MapReduce[2]和彈性分布式數據集（RDD,resilient distributed dataset）[3]編程模型為代表的批處理[4]模式已經無法滿足應用的實時性要求，大數據流式計算應運而生。流式計算將計算任務抽象為數據流模型，具有實時性、易失性、突發性、無序性和無限性的特征[5]，能夠提供大規模數據的分布式實時處理，已經分別得到學術界和產業界的廣泛關注。其中，Apache Flink 是目前應用最廣泛的新興流式計算平臺。然而，面對不斷波動變化的計算負載，Flink平臺存在可擴展性和可伸縮性較差，應對負載波動的能力不足，已經成為制約平臺發展的嚴峻而又亟待解決的問題，受到開源社區的重點關注。

針對上述問題，本文提出基于流網絡的Flink平臺彈性資源調度策略（FAR-Flink,Flow-network based auto rescale strategy for Flink），并將其應用于Flink 平臺。本文的主要貢獻如下。

1)提出基于流網絡對流式計算拓撲進行建模的思想，并在此基礎上提出流網絡劃分等概念，定義了上下游節點的劃分方式，為提出彈性資源調度算法和彈性資源調度策略提供了模型的支撐。

2)提出流網絡的容量值構建算法，采用初值定義和反饋調節相結合的方式，確定流網絡每條邊的容量值，為保障彈性資源調度算法的優化效果奠定基礎。

3)在流網絡模型的基礎上提出彈性資源調度算法，在合理分配新增計算負載的同時定位集群性能瓶頸，生成彈性資源調度計劃，通過擴充計算節點突破瓶頸，提高集群整體性能。

4)提出一種狀態數據分簇和分桶管理的思想，并在此基礎上提出狀態數據高效遷移策略，有效提高了集群執行檢查點和數據恢復的效率，降低執行彈性資源調度計劃的時間開銷。

2 相關研究

面對大數據流式計算平臺存在突發性[6]問題，現有的流式計算平臺均未提供默認的彈性資源調度機制，計算平臺存在可伸縮性和可擴展性不足的問題，已經得到學術界和產業界的廣泛關注。目前，現有的研究成果主要基于以下4 個出發點來解決該問題：1)通過彈性資源調度策略提高集群的可伸縮性；2)通過優化任務調度策略提高集群的性能；3)通過優化數據分配策略提高作業的執行效率；4)通過減小網絡通信開銷以降低計算時延。

通過彈性資源調度策略提高集群資源的可伸縮性，是提高集群性能最有效的方法。其中，文獻[6]提出適用于Nephel[7]數據流處理平臺的響應式資源調度策略，通過建立數學模型計算每個算子的并行度，并通過任務遷移實現集群資源的動態伸縮，但其任務遷移過程中的網絡傳輸開銷較大。文獻[8]通過監控集群的性能指標，建立針對Storm 平臺無狀態數據流的彈性資源調度策略。文獻[9]通過提出分布式彈性資源管理協議，實現集群規模對輸入負載的快速響應。文獻[10]通過提出高效的狀態數據管理策略，實現集群在橫向和縱向上的可擴展性。文獻[11]提出基于動態參數優化的彈性資源數據流處理平臺。此外，文獻[12-16]分別從資源分配、任務調度、負載優化等不同的角度，提出數據流彈性資源調度策略。

優化任務調度策略是另一種提高集群性能的有效方法。文獻[17]提出優化流式作業的通用原則，實現了透明的拓撲優化策略。文獻[18]提出對流式計算拓撲的圖優化策略，在提高吞吐量的同時降低時延。文獻[19]通過對關鍵路徑節點進行重分配，在降低計算時延的同時減少系統能耗。此外，文獻[20]通過優化任務拓撲結構的方式，有效提高了集群性能。

優化數據分配策略是提高作業執行效率的有效方式，其中最典型的策略是負載均衡。文獻[21]通過實現上下游節點算子的靈活遷移和動態鏈接[22]，應對內存不足造成的背壓（backpressure）問題。文獻[23]提出自定義的代價模型，在鄰近代價閾值時啟動分區映射算法[24]，實現節點間計算負載的最優分配。文獻[25]將流式計算拓撲定義為流網絡模型并從中尋找優化路徑，從而提高集群吞吐量。此外，文獻[26-27]也分別提出不同的負載均衡策略，以提升集群的整體性能。

通過降低網絡傳輸開銷以提高傳輸性能，也是提高作業執行效率的有效方式。文獻[28]提出根據計算任務的不同類型動態調節緩沖池的分區大小，有效降低傳輸開銷以提高傳輸性能。文獻[22]通過對并行度相等的上下游算子進行動態鏈接，有效減少了節點間的數據通信。文獻[29]通過將節點內進程間網絡通信轉化為線程間通信，有效降低了通信開銷。此外，文獻[30-31]分別從不同角度降低數據傳輸的通信開銷，提高集群性能。

其中，通過彈性資源調度策略提高集群的可伸縮性，能夠從根本上解決負載波動導致集群響應能力不足的問題，但現有研究成果多針對Storm 平臺提出，受平臺內核之間的差異而無法直接移植于Flink。因此本文提出適用于Flink 平臺的彈性資源調度策略，與現有研究成果的不同之處介紹如下。

1)基于流網絡對集群拓撲結構進行抽象，采用初值定義和反饋調節相結合的方式構建模型，準確評估了節點的計算能力和集群的性能瓶頸，并在此基礎上制定了最優的彈性資源調度計劃。

2)重點考慮有狀態數據流的任務調度問題，提出狀態數據分桶管理的思想，有效提高了狀態數據的遷移效率。

3)通過分析Flink 平臺內核的結構和特點，結合狀態數據管理和檢查點機制，克服平臺內核之間的差異，提出了適用于Flink 的彈性資源調度策略。

4)在實驗中，提出通過監控數據傳輸量評估數據遷移效率的方法，驗證了狀態數據遷移算法的優化效果。在此基礎上，通過對比實驗得出主流彈性資源調度策略的優缺點及其適用場景。

3 問題建模與分析

本節從有狀態流式計算的特點出發，建立了流網絡模型和狀態數據管理模型。首先通過分析有向無環圖（DAG,directed acyclic graph）中節點容量與流量的數學關系，建立流網絡模型，為彈性資源調度算法提供了模型的支撐。此外，通過分析現有狀態數據管理策略的不足，指出影響狀態數據遷移效率的主要原因，為設計高效的狀態數據遷移算法提供了理論依據。

3.1 流網絡模型

在分布式數據流處理平臺中，用戶定義的計算邏輯是封裝在算子中的，并由各工作節點并行化執行算子的計算邏輯，各工作節點被稱為對應算子的實例。由此可知，當集群的計算資源總量和拓撲結構確定后，其所能承擔的最高計算負載也隨之確定，計算資源不足將導致集群性能急劇下降。

因此，為了解決集群資源不足導致的性能問題，將流式計算的拓撲結構定義為流網絡模型。在該模型中，將節點能夠處理的最高計算負載定義為容量c(vi,vj)，當前時刻節點實際處理的負載定義為流量f(vi,vj)，通過構建對應的增進網絡并尋找優化路徑，實現計算負載的最優分配，從而定位集群的性能瓶頸并制定相應的彈性資源調度計劃。

定義1流網絡。如圖1 所示，設集群拓撲為單源有向無環圖G=(V,E)，其中V={v1,v2,…,vn}是圖中所有節點的集合，s∈S是流網絡的源點，ti∈T是匯點，E={(vi,vj)|i,j∈[1,n],n=|V|}是有向邊的集合，(vi,vj)表示節點vi與vj間的傳輸鏈路。且?(vi,vj)∈E有c(vi,vj)≥0，表示邊(vi,vj)允許數據傳輸速率的最大值，稱為邊(vi,vj)的容量；f(vi,vj)表示邊(vi,vj)當前時刻的實際傳輸速率，稱為邊(vi,vj)當前的流量。此外，當前時刻從數據源點向網絡輸入數據速率的和記為該網絡的流量，即

圖1 流網絡模型

在流網絡中，容量與流量的單位均為tuple/s，且有0≤f(vi,vj)≤c(vi,vj)恒成立。

通過定義流網絡模型，將流式計算的DAG 拓撲轉化為單源有向無環圖，用容量c(vi,vj)表示邊(vi,vj)允許數據傳輸的最大值，并用流量f(vi,vj)表示邊(vi,vj)實際的傳輸速率。其中，每條邊的容量值表示對應節點的處理能力和數據鏈路的傳輸能力，只有準確計算每條邊的容量值才能有效評估集群的性能指標，從而實現集群資源的最優分配。因此，通過4.1 節提出的流網絡構建算法，采用數學計算與反饋調節相結合的方式，計算網絡中每條邊的容量大小，構建整個流網絡的形態，對彈性資源調度算法的執行是至關重要的。

定義2增進網絡。設有流網絡G=(V,E)，且f是G的一個流，則Gf=(Vf,Ef)是流網絡G由f產生的增進網絡，其形態如圖2 所示，其中?vi∈V有vi∈Vf，?(vi,vj)∈E在Gf中對應的增進容量函數cf(vi,vj)為

其中，c(vi,vj)為原網絡中邊的容量函數，f(vi,vj)為原網絡中邊的流量函數

由定義2 可知，增進網絡表示在原流網絡中，每條邊上流量提升的空間。因此在式(2)中，對于所有與原網絡同向的邊，其容量值為原網絡中容量與流量的差值；對于所有與原網絡反向的邊，其容量值與原網絡的流量值相等。這樣，在增進網絡中尋找一條從源點到匯點的無環路徑，表示原流網絡中提升流量的空間，即提升集群吞吐量的空間。

定義3優化路徑。設有流網絡G=(V,E)，Gf=(Vf,Ef)是G由流f產生的增進網絡。則在Gf中從源點s至任意匯點ti的一條無環路徑p:s→vi→ti都是原網絡G的一條優化路徑。其中，集合P={(vi,vj)| 邊(vi,vj)在路徑p上}為優化路徑上邊的集合，則該路徑對應的遞增量為

其中，cf(vi,vj)為邊(vi,vj)在增進網絡中的容量值。因此，將fp作用于f，得到流f在路徑p上的一個遞增流，記為f'=f↑fp，且對應邊的流量為

其中，f(vi,vj)是原網絡中對應邊的流量。由于遞增量表示優化路徑上每條邊均能夠提升流量的空間，因此選取該路徑上增進容量的最小值。

設流網絡G=(V,E)的一個流為f，Gf=(Vf,Ef)為G由流f產生的增進網絡，路徑p為Gf中的任意一條優化路徑，則流f在路徑p上的遞增流f↑fp也是原網絡G的一個流，且其流量為

圖2 增進網絡模型

由定義2 和定義3 可知，增進網絡描述了在原網絡中，每條邊在容量限制下可能提升流量的空間，則優化路徑表示提升網絡流量、合理分配堆積數據的有效方案。因此，當源點產生堆積數據時，沿著優化路徑的方向分配負載，就一定能夠提高集群的吞吐量。當增進網絡中不存在優化路徑時，集群達到其吞吐量的峰值，則一定存在一個算子成為整個集群的性能瓶頸。

3.2 資源優化配置模型

根據流網絡模型可知，通過準確計算每條邊上容量與流量的取值，量化集群計算能力與計算負載的數學關系，從而分析集群當前時刻的資源分配和使用情況。然而，當集群因計算資源不足而導致性能下降時，需要定義資源優化配置模型，通過定義流網絡的劃分來尋找集群的性能瓶頸，從而制定最優的彈性資源調度計劃。

定義4流網絡劃分。設有流網絡G=(V,E)，其中s∈S是流網絡的源點，ti∈T是匯點。則該網絡的一個劃分D=(X,Y)將節點集V分為X和Y這2個集合，其中Y=V-X，使s∈X，ti∈Y，且有X∩Y=?，X∪Y=V。 ?vi,v j∈Oi有vi,v j∈X或vi,v j∈Y。則該劃分D=(X,Y)對應的容量記為

該劃分對應的流量記為

其中，容量最小的劃分為該流網絡的最小劃分。

如圖3 所示，流網絡的一個劃分將數據源點和匯點分在2 個不同的集合中，且同一個算子的不同實例不橫跨任何一個劃分。因此，一個劃分中容量和流量的關系反映了不同算子之間在資源配置和計算性能上的差異，為定位整個集群的性能瓶頸提供了可行的方案。

設流網絡G=(V,E)的一個流為f，Gf=(Vf,Ef)是G由f產生的增進網絡。當Gf中不存在任何優化路徑時，f是G的最大流，則至少存在一個劃分D=(X,Y)，使且D是G的最小劃分。

當增進網絡中不存在任何優化路徑時，原網絡的流量達到最大值，且當前流量值與最小劃分的容量值相等，則最小劃分所對應的算子成為集群的性能瓶頸。因此，FAR-Flink 策略的核心思想為：首先，通過計算DAG 模型中每條邊的容量大小，將其轉化為流網絡模型；其次，通過計算對應的增進網絡和優化路徑，實現計算負載的最優分配，當集群性能達到瓶頸時，再通過尋找流網絡的最小劃分，制定彈性資源調度計劃；最后，通過基于分簇和分桶的狀態數據管理模型，實現狀態數據高效遷移算法，并完成彈性資源調度計劃的執行。

3.3 狀態數據管理模型

根據資源優化配置模型，當數據源點的輸入速率發生變化時，可制定最優的彈性資源調度策略。但由于Flink 是有狀態數據流處理平臺，每個節點會保存當前的狀態數據，為了實現高效的節點間狀態數據遷移策略，降低彈性資源調度計劃的執行開銷，必須建立合理的狀態數據管理模型，并在此基礎上設計高效的數據遷移算法。在數據遷移過程中，盡可能降低對Hadoop 分布式文件系統（HDFS,Hadoop distributed file system）的訪問頻次，減少不必要的數據傳輸，從而有效提高動態資源調度策略的執行效率。

圖3 流網絡劃分示意

如圖4 所示，當集群由v1,v2∈O1這2 個實例擴充至v1,v2,v6∈O1這3 個實例時，需要考慮原本由2 個節點維護的狀態數據d1和d2如何分配到3 個節點中，以實現節點間計算任務的最優分配；此外，在對狀態數據執行快照和重分配的過程中，都會頻繁地訪問HDFS 執行數據讀寫的操作，這會占用大量的網絡傳輸資源，進而影響節點間的數據傳輸性能。針對上述問題，提出高效的節點間狀態數據遷移算法，能夠有效提高動態資源調度計劃的執行效率。

圖4 狀態數據分配模型

設算子On待處理的數據為二元組tuplei=(key,value)，則該元組對應的簇為

即具有相同key 的hash 值的數據元組屬于同一個簇。在現有的Flink 平臺中，目前是按key 的hash值不同，分簇對狀態數據進行管理的，稱為KeyedState，是目前狀態數據管理的主要策略。但由于數據元組有多種不同的key，hash 函數發生碰撞的概率較低，通常狀態數據的分布較為分散，在數據遷移過程中需要頻繁訪問HDFS，占用大量的網絡傳輸資源并引入較高的傳輸開銷，進而影響了彈性資源調度算法的執行效率。因此，通過提出狀態數據分桶管理的思想，并設計優化的狀態數據遷移算法，能夠有效減少節點對HDFS 的訪問頻次，減少網絡傳輸開銷以提高算法性能。

4 動態資源調度策略

在第3 節建立的相關模型基礎上，本節提出了FAR-Flink 策略，該策略分別包含流網絡構建算法、彈性資源調度算法和狀態數據遷移算法。該策略主要分為以下3 個步驟，具體的執行流程如圖5 所示。

步驟1通過流網絡構建算法計算每條邊的容量大小，建立流網絡模型。

步驟2通過彈性資源調度算法定位性能瓶頸，制定調度策略。

步驟3通過狀態數據遷移算法執行調度策略，實現集群規模的彈性伸縮。

圖5 FAR-Flink 策略執行流程

4.1 流網絡構建算法

根據流網絡模型可知，準確評估網絡每條邊的容量大小對策略的執行效果是至關重要的。經實驗分析得出，容量值與以下4 個因素有關。

1)節點間的網絡傳輸性能和傳輸資源占用情況，傳輸資源不足將導致容量值減小。

2)節點內的計算性能和計算資源占用情況，計算資源不足導致容量值減小。

3)節點所承擔計算任務本身的復雜程度，計算任務越復雜則容量值越小。

4)下游節點的計算資源剩余情況，根據Netty組件的水位機制，下游節點的數據阻塞會引起反壓現象，導致上游節點容量值下降。

在流式計算平臺中，數據傳輸性能對計算的實時性影響較大，因網絡擁塞導致數據在節點緩存中被滯留是流式計算平臺面臨的主要性能瓶頸。當一臺PC 機中關閉所有與Flink 無關的進程時，在純凈的網絡環境中，該節點實際可用于數據傳輸的網絡帶寬資源為

其中，Nvi(Data)為實際可用于節點間數據傳輸的網絡帶寬資源；為物理節點間原有的網絡帶寬資源總量，如節點間使用百兆帶寬的網絡連接，則為操作系統本身的靜態網絡傳輸開銷；為進程間心跳信息的傳輸開銷；為節點間歇性執行檢查點（checkpoint）時向HDFS 和Zookepper寫入狀態數據的傳輸開銷；為其他隨機因素可能產生的極少量傳輸開銷。單位均為MB/s。

因此，該節點對應輸入鏈路的容量值為

其中，|Eji|為節點vi對應輸入邊的數目，size(tuple.fi)為數據元組中每個字段所占空間的大小，單位為B。

根據上述分析，流網絡構建算法的執行過程如算法1 所示。

算法1流網絡構建算法

輸入集群拓撲結構T，最高計算時延閾值θ

輸出流網絡G

在算法1 中，首先確定流網絡中頂點和邊的集合（步驟1)和步驟2)）；其次根據每個計算節點的網絡傳輸資源，計算節點對應輸入邊的初始容量值（步驟3)～步驟6)）；最后在作業執行過程中，根據計算時延的平均值與閾值之間的關系，對流網絡每條邊的容量值進行反饋調節。當平均計算時延大于設定的閾值，且對應邊的流量值仍小于容量時，表明設定的容量過大，則以η為步長減小對應邊的容量；當平均計算時延遠小于設定的閾值，但流量值已經接近容量時，表明對應邊的容量值過小，則以η為步長增大對應邊的容量。實驗數據表明，通過網絡開銷計算出的初始容量值往往是數據傳輸的極限值，受計算開銷等其他因素的影響，實際合理的容量值應小于初始值，因此需要通過反饋調節將容量值調整至合理的范圍內。

4.2 彈性資源調度算法

通過流網絡構建算法建立流網絡模型后，當源點產生數據堆積時，通過彈性資源調度算法合理分配計算負載、定位性能瓶頸并生成彈性資源調度計劃。

圖6 為彈性資源調度策略示意，假設圖中有劃分D=(X,Y)，當且僅當

則Y集合中第一個算子O2有可能成為集群的性能瓶頸，需要增加資源并擴大O2的并行度，其中λ是當前劃分D=(X,Y)對應流量與容量的比值，且有0.85≤λ≤1，即當一個劃分對應的流量達到容量值的85%時，算法認為該劃分對應的算子可能成為集群的性能瓶頸。

圖6 彈性資源調度策略示意

基于資源優化配置模型，彈性資源調度算法的執行過程如算法2 所示。

算法2彈性資源調度算法

輸入流網絡G，流網絡的一個流f，數據源點每個分區的堆積值lags[]

輸出完成任務調度后新的流網絡模型G′

在算法2 中，根據增進網絡與優化路徑的定義，首先由流網絡G的一個流f生成對應的增進網絡Gf（步驟1)～步驟5)），當數據源點產生堆積時，在增進網絡中尋找優化路徑，并沿其方向提高原網絡的流量（步驟6)～步驟11)）。當增進網絡中不存在優化路徑時，如果數據源點仍有堆積，則在流網絡中尋找流量與容量的比值大于λ（即滿足式(10)）的劃分，增加其Y集合中第一個算子的并行度（步驟12)～步驟18)），從而得到需要擴大并行度的算子。最后，通過調用狀態數據遷移算法（步驟19)），完成節點間的狀態數據遷移，從而實現計算資源的動態調度。

4.3 狀態數據遷移算法

在執行彈性資源調度策略時，節點間狀態數據遷移會產生大量網絡傳輸開銷，但Flink 狀態數據管理策略并不適用于高效的數據遷移。因此，通過提出狀態數據管理模型和遷移算法，降低數據傳輸開銷，提高遷移效率。

定義5桶。設算子On待處理的數據為一個二元組tuplei=(key,value)，算子On共持有kn個桶，則該元組對應的桶為

則對應負責處理該元組的實例為

其中，vi為算子On的第i個實例，。這樣就建立了數據桶與計算節點間的映射關系，將桶作為計算節點維護狀態數據的基本單位。此外，參數kn是算子On所持有桶的數目，用戶可根據狀態數據的規模和復雜程度進行適當的調整。

圖7 表示由狀態數據簇到分桶管理的映射方式，其中kn=10，算子并行度由|On|=3 擴大至|On|=4。在狀態數據遷移算法中，結合Flink 支持的檢查點（checkpoint）機制，在執行快照時將狀態數據以桶為單位發送至HDFS，同一個桶的數據存儲在相鄰的位置。當算子的并行度改變時，根據式(11)和式(12)重新計算每個桶到節點的映射關系，最后由節點從HDFS 中拉取對應的狀態數據，完成計算節點的狀態數據恢復。

結合狀態數據分桶管理的特點，可提出狀態數據遷移算法如算法3 所示。

算法3狀態數據遷移算法

輸入需增加并行度的算子集 operator[]

輸出完成任務調度后新的流網絡模型G′

圖7 狀態數據遷移示意

在算法3 中，首先對每個節點執行檢查點操作（步驟3)～步驟7)）：將節點的狀態數據發送至HDFS，并將狀態數據的句柄信息保存在ZooKeeper中，記錄狀態數據對應的桶。其次從資源池中獲取一個新的計算節點，擴大對應算子的并行度，并由JobManager 重新計算狀態數據桶到計算節點的映射關系。再次根據新生成的集群拓撲結果，對每個節點執行數據恢復操作：從ZooKeeper 中獲取狀態數據對應的句柄，并從HDFS 中拉取對應的數據。最后再次執行算法1，以修正每條邊對應的容量值。

4.4 相關參數分析

在流網絡構建算法中，參數η是動態調節容量值的步長。η過大會導致調整幅度過于劇烈，η過小會導致調整幅度不夠，2 種情況均會導致最終的容量值不準確。因此參數η的計算方法如下。

根據定義1，c(vi,vj)是邊(vi,vj)每秒鐘允許傳輸的最大元組個數，且1 s=1 000 ms。根據文獻[26]的時延統計算法，節點間傳輸數據的平均計算時延為

其中，vj.fi是數據元組到達節點vj的發現時間[26]，vj.di是數據元組離開節點vj的完成時間[26]。則容量值為對應節點在1 000 ms內所能傳輸的平均元組數目，將式(13)代入得

由于η是調節容量值的步長，即容量函數的變化率，則對容量函數求導得

此外，根據容量值與計算時延的函數關系，通過記錄的平均計算時延，求得調節前與調節后的容量值偏差應為

因此，為了避免調整過于劇烈導致容量值出現抖動的現象，學習參數η取容量函數的導數與偏差的較小者，即

實驗表明，通過式(17)分別計算流網絡中每條邊學習因子η，再通過算法1 調整對應邊的容量函數，能夠取得比較準確的容量值，流網絡構建算法有比較好的優化效果。

4.5 算法性能分析

由于流式計算對平臺的性能和實時性有很高的要求，因此彈性資源調度策略應具備很高的性能，且不能引入過高的時間開銷。流網絡構建算法首先遍歷流網絡的每條邊，以確定其容量的初始值；再遍歷每個節點，從而對邊的容量值進行反饋調節，因此構建算法的時間復雜度為T(n)=O(|V|+|E|)。

此外，彈性資源調度算法是基于廣度優先搜索（BFS,breadth first search）實現的流量遞增算法，已知BFS 算法的時間復雜度為T(n)=O(|V|+|E|)，且在算法2 中，3 次循環的最高循環次數分別為流網絡的節點數目、邊的數目以及當前流量與彈性資源調度量的差值，因此彈性資源調度算法的時間復雜度為T(n)=O(|V|+(|V|+|E|)(fmax-f)+|D|)，由于流網絡中邊的數目大于節點數目和劃分的數目，因此算法的時間復雜度為T(n)=O(|E|(fmax-f))。

數據遷移算法是典型的分布式并行化算法，其中的每個步驟都分別在不同的節點上執行，節點之間通過ZooKeeper 進行統一協調，并且計算過程較為簡單，其時間開銷主要取決于狀態數據的規模和節點間網絡傳輸的性能，因此沒有具體的數學計算式來表達其時間復雜度，本文5.4 節通過實驗驗證了算法的時間開銷較原系統有一定的下降。

5 實驗與分析

為了驗證FAR-Flink 策略的有效性，本文通過設置對比實驗，將FAR-Flink 與原生Flink 1.6.0 版本的系統形成對比，并與本文工作密切相關的Elastic Nephel[6]策略形成對比，通過執行代表不同作業類型的典型Benchmark，驗證了算法的優化效果和執行開銷，并分析了相關參數對算法的影響。

5.1 實驗環境

實驗搭建的集群由21 臺同構的PC 機組成。其中，Flink 集群包含一個JobManager 節點、6 個TaskManager 節點，資源池中包含4 個TaskManager備用節點，在執行彈性資源調度計劃動態擴展資源時啟動備用節點，并部署相應的計算任務。其他相關組件包括3 個節點構成的Hadoop 集群、3 個節點構成的Kafka 集群和3 個節點構成的ZooKeeper 集群。另外，由一個獨立的節點執行流網絡構建算法和彈性資源調度算法。其中，每個同構節點的硬件配置和軟件配置分別如表1 和表2 所示。

表1 節點硬件配置參數

表2 節點軟件配置參數

同時，為了使Flink 集群達到最優的計算性能，根據現有的軟硬件環境，對Flink 的相關配置參數進行了調整，其中重要的配置項及其參數值如表3所示。

表3 性能參數配置

5.2 構建算法性能測試

在流網絡模型中，每條邊的容量值應客觀反映節點的計算能力和節點間的傳輸能力，容量值評估不準確可能會導致計算時延升高而無法滿足計算的實時性要求。為了探究容量函數與計算時延的關系，實驗為每條邊設置相等的容量值，并不斷遞增。同時，基于Flink Metrics 體系的Latency Tracking機制采集的計算時延如圖8 所示。

圖8 容量與計算時延關系

由圖8 可以看出，隨著時間的推移，容量值持續增大，每個TaskSlot 的計算時延均有一定的波動，但總體呈上升趨勢，甚至出現階躍上升的現象，且計算時延越高，其波動變化越劇烈。這表明過高地估計節點的計算能力，會導致數據元組被阻塞、計算時延升高，但在滿足計算時延約束的前提下容量值總會被盡可能地增加，從而提高集群的吞吐量。因此，在當前實驗環境中，當容量函數取30 000～40 000 tuple/s 時，平均計算時延在400 ms 以下，這是比較合理的取值區間。

此外，在流網絡構建算法中，參數η確定了調節容量函數的步長，對流網絡的穩定性和收斂速率有決定性作用。如圖9 所示，當使用固定的η取值執行構建算法時，流網絡中存在收斂速率和穩定性難以權衡的問題：較大的η值會導致容量值劇烈波動而無法趨于穩定，較小的η值會導致網絡收斂時間較長（約50～60 s）。但通過式(17)計算的動態η取值能夠客觀反映當前容量值與理想值的差，隨著η的不斷減小，流網絡的容量值迅速收斂于理想值，并逐漸趨于穩定。

圖9 學習參數對比

5.3 FAR-Flink 性能測試

文獻[6]提出針對Flink 平臺內核的彈性資源調度策略，與本文工作的研究目標最接近。為了驗證FAR-Flink 的優化效果，實驗在Flink 原系統、Elastic Nephel[6]和FAR-Flink 上分別執行WordCount、TwitterSentiment、IncrementalLearing 和Streaming-Benchmarks 代表不同作業類型的標準Benchmark，它們分別來自Flink 源碼中的示例和Intel 等公司在開源社區Github 上的貢獻。實驗結果分別如圖10～圖13 所示。

采用與5.2 節相同的實驗配置，分別在原系統和FAR-Flink 上執行WordCount 作業，并采集Kafka緩沖池中的數據堆積和計算時延，如圖10 所示。由圖10(a)和圖10(b)可知，計算時延與Kafka 中的數據堆積呈正相關，即實驗驗證了數據堆積導致時延升高的理論推測是正確的。此外，由于原系統執行默認的資源調度策略，因資源不足導致計算時延不斷升高。FAR-Flink 通過彈性資源調度算法合理分配Kafka 中的堆積數據，并動態增加了算子OFlatMap的并行度。但由于執行狀態數據遷移有一定的時間開銷，導致數據堆積有短暫的上升。從圖10(c)中可以看出，Elastic Nephel(EN)的時延上升時間較長。FAR-Flink 執行任務遷移過程中，持續時間比EN 縮短了約40 s。

圖10 WordCount 作業執行效率對比

TwitterSentiment 是在生產環境中實際應用的標準Benchmark。實驗采用與文獻[32]相同的實驗配置，并以10 s 為周期采集節點的內存使用率。得到如圖11 所示的實驗結果。

由圖11 可知，隨著計算負載的不斷升高，EN和FAR-Flink 分別在第540 s 和第720 s 各增加了一個節點，擴大算子并行度并執行狀態數據遷移。但由于作業執行中產生狀態數據并占用大量內存資源，狀態數據遷移過程產生了一定的時間開銷：EN系統的2 次遷移分別消耗約14 s 和18 s，FAR-Flink分別消耗約7 s 和13 s。但FAR-Flink 遷移過程中吞吐量較低且內存資源消耗較高，這是因為執行restore 操作需要一定時間，且遷移過程中產生大量的HDFS 寫操作。但隨著計算資源的增加，系統的吞吐量有明顯的上升，基本滿足輸入負載的要求，且內存資源占用降低至合理可接受的范圍內，這說明2 種彈性資源調度策略均有效提升集群性能。

圖11 TwitterSentiment 作業執行情況對比

為了驗證FAR-Flink 在高計算復雜度、高CPU資源占用場景下的優化效果，實驗運行了IncrementalLearing 作業，并以10 s 為周期采集節點的CPU 利用率，得到如圖12 所示的實驗結果。

圖12 IncrementalLearing 作業執行情況

由圖12 可知，集群執行彈性資源調度策略計劃的過程需要一定的時間開銷，EN 和FAR-Flink都在第600 s 左右增加了一個計算節點，其中EN執行策略的時間開銷較高（約42 s），但執行過程中其他節點計算性能下降，集群吞吐量驟減。FAR-Flink 執行策略的時間開銷較低（約18 s），但執行restore 過程中整個集群有極短暫的停滯，其中第600 s 時檢測輸出數據量為0 tuple/s，在短暫的時間內恢復了任務所有計算節點的狀態數據，因此在下一個階段的吞吐量值急劇回升。在資源利用方面，EN 在數據遷移過程中的CPU 利用率值劇烈抖動，FAR-Flink 的CPU 利用率值急劇下降后又快速回升，這都與其執行調度策略的過程相關，與吞吐量的檢測結果是一致的。總體上，2 種策略均通過增加計算資源提高了集群性能，FAR-Flink 有效縮短了EN 執行調度策略的時間，但其執行過程中計算任務有非常短暫的停滯。

為了進一步驗證FAR-Flink 在實際應用場景下的性能，實驗采用Yahoo 公司在GitHub 上開源的Streaming-Benchmarks[32]，并分別從吞吐量、計算時延、Kafka 數據堆積、堆內存利用率、CPU 利用率5 個不同的維度，監測集群和作業的運行情況，監測結果如圖13 所示。

由圖13 可以看出，隨著計算負載的持續上升，EN 和FAR-Flink 系統都分別出現資源不足導致集群性能下降的問題。其中，EN 系統連續動態增加了2 個計算節點，由于計算復雜且節點狀態數據規模較大，數據遷移過程產生了較高的時間開銷，在第一次數據遷移后集群吞吐量有少量的回升就立刻進入第二次遷移，整個遷移過程共持續約273 s。FAR-Flink 首先通過彈性資源調度算法合理分配計算負載，并在第600 s 和第780 s 時分別動態增加一個計算節點，其數據遷移過程分別持續了21.6 s和36.7 s，較EN 系統的數據遷移時間有明顯的下降。同時，2 種算法均通過動態增加計算資源有效提升了集群性能，其中EN 系統的數據遷移時間較長，但遷移過程中集群可保證較低的吞吐量。FAR-Flink 系統能夠在前期合理分配計算負載，且有效縮短了數據遷移過程的時間開銷，但遷移過程中作業有極短暫的停滯，遷移完成后集群性能有較明顯的上升。

綜上所述，實驗在4 種資源調度策略下分別執行不同的標準Benchmark，通過性能對比得出了不同算法的優缺點及適用場景，實驗結果如表4 所示。實驗證明，FAR-Flink 通過合理分配計算負載、動態增加計算資源、降低數據遷移開銷3種策略相結合的方式，有效提高集群性能。與原系統相比，在計算負載波動上升期間，針對不同類型的Benchmark，集群吞吐量平均提高了27.61%，狀態數據遷移時間平均縮短了45.36%，具有一定的優化效果。

5.4 數據遷移開銷測試

圖13 Streaming-Benchmarks 作業執行效率

表4 對比實驗結果

為了準確評估FAR-Flink 執行數據遷移過程產生的網絡傳輸開銷，驗證數據遷移不會長時間占用過多的網絡傳輸資源而影響集群性能，實驗分別執行TwitterSentiment 和Streaming-Benchmarks 作業，并監測網絡傳輸數據以評估網絡傳輸開銷，得到如圖14 所示的實驗結果。

由圖14 可知，除去節點間可忽略的靜態數據傳輸外，計算節點會依據相關配置周期性地執行狀態數據快照，并向HDFS 發送數據，在執行動態資源調度計劃時，節點會從HDFS 拉取相應的數據并執行狀態數據的恢復操作。在2 種策略執行過程中，需要傳輸的數據總量是基本相同的。但由于EN 以塊（bulk）為單位從遠端拉取數據，其數據分布較為分散且存在大量碎片化數據，因此單位時間內的數據傳輸速率較低，傳輸時間較長。FAR-Flink 以桶（bucket）為單位從遠端拉取數據，其數據分布較為集中且幾乎沒有碎片化數據，計算節點在短時間內集中拉取需要的數據，數據傳輸速率較高，傳輸時間較短。

綜上所述，通過基于分簇和分桶的狀態數據遷移算法，合理應對碎片化數據傳輸的問題，有效降低數據遷移的網絡傳輸開銷，通過提高傳輸效率縮短執行動態資源調度策略的時間。但實驗結果表明，這種方式仍會產生一定的時間開銷，在合理可接受的范圍內對執行效率有一定的影響，如何進一步提高數據遷移效率以縮短遷移時間，將是下一步研究工作的主要方向。

圖14 數據遷移的網絡傳輸開銷對比

6 結束語

作為集流處理與批處理為一體的統一大數據處理平臺，Apache Flink 得到學術界和產業界的廣泛關注，但其可擴展性和可伸縮性不足的問題，已經嚴重制約了平臺的發展。本文提出了基于流網絡的數據流動態資源調度策略，通過合理分配負載、動態增加資源、高效數據遷移3 種策略相結合的方式，從根本上解決了因計算資源不足而影響集群性能的問題，并有效降低了網絡傳輸的時間開銷。

但本文算法也存在一定的局限性，首先，本文算法對ZooKeeper有很強的依賴性，需要ZooKeeper中保存相關的數據結構；其次，狀態數據遷移過程中作業會有極短暫的停滯（約2～3 s），但算法的執行開銷在可接受的范圍內。因此，未來的研究工作將主要集中于以下3 個方面。

1)降低資源調度策略對ZooKeeper 的依賴程度，在限制單點計算和傳輸負載的前提下，嘗試由JobManager 統一提供資源調度和數據共享服務。

2)通過提出新的狀態數據管理和遷移策略，降低數據遷移開銷，縮短調度計劃的執行時間，提高執行效率。

3)通過提出計算任務的熱部署方案和節點間狀態數據實時共享策略，實現對用戶作業無感知的、完全透明的在線資源調度策略。