基于流網絡的流式計算動態任務調度策略

李梓楊，于 炯,，卞 琛，魯 亮，蒲勇霖

(1.新疆大學 信息科學與工程學院，烏魯木齊 830046； 2.新疆大學 軟件學院，烏魯木齊 830008)

0 引言

隨著互聯網技術和信息產業的不斷發展，全球數據量呈幾何式增長，截止2015年全球數據總量達8.61 ZB，并預計到2020年全球數據總量將超過40 ZB[1]，同時，通過移動互聯、社交媒體、全球定位系統(Global Positioning System, GPS)導航等新的服務模式，大數據[2]產業及相關服務已經深入到人們生活的方方面面，也為互聯網企業帶來巨大收益。然而隨著數據價值的時效性變得越來越明顯，集群必須以毫秒級的延遲從大規模數據中提煉出有價值的信息，才能滿足用戶對數據分析的實時性要求，大數據流式計算[3]應運而生。流式計算具有實時性、易失性、無序性、無限性和突發性的特征[4]，能夠提供高效的數據分析服務，已在交通預警、實時推薦等對實時性要求高的場景中得到廣泛應用；但流式計算的技術發展也面臨著一些挑戰，多樣的輸入數據源和不斷變化的輸入數據速率對集群的負載承受能力和可伸縮性提出了更高的要求，特別是輸入速率的急劇上升會給集群造成很大的負載壓力，如果應對不力就會造成數據元組被阻塞或丟棄，甚至出現節點崩潰等現象，影響計算的實時性和準確性。

流式計算的發展誕生了不同特點的數據流處理平臺，Apache Flink[5-9]是新興的目前產業界應用最廣泛的平臺之一。與Storm[10]平臺相比，Flink能提供Exactly-Once的可靠性計算[11]以及更完善的背壓機制[12]，并支持用戶定義的時間窗口[13]，但在輸入速率上升階段的吞吐量仍有待提高，因此，本文提出基于流網絡模型的動態任務調度(Flow Network based Dynamic Dispatching, FNDD)策略。該策略將流式計算拓撲轉化為流網絡模型，通過容量檢測算法和最大流算法實現流式計算平臺的動態任務調度。經實驗驗證得出，該策略對不同作業類型的優化效果有較明顯的區別：其中集群在WordCount作業中的吞吐量平均提高了29.41%，在TwitterSentiment作業中的吞吐量平均提高了16.12%，在TeraSort作業中的吞吐量平均提高了38.29%。

1 相關工作

為了解決流式計算中輸入速率急劇上升導致數據元組被阻塞或丟棄，進而影響計算的實時性和準確性的問題，必須提出一種在輸入速率上升階段的任務調度策略，使其能夠根據節點的處理能力合理地負載分配，并根據實際情況動態變化，從而在保證低延遲的同時提高吞吐量。

針對輸入數據的速率急劇上升導致集群的負載壓力增大的問題，現有的研究成果大多只關注節點內的計算開銷而忽略了節點間的傳輸開銷，且大多不適用于Flink平臺。文獻[14]研究發現：集群拓撲結構和節點內緩存大小對任務的計算延遲和吞吐量有較大的影響，提出通過調整緩沖區的大小以及動態鏈接(Chain)部分算子的思想，在滿足計算延遲約束的前提下盡可能提高吞吐量；但其同步的性能監控策略產生了較大的時間開銷，導致該算法不能應用于大規模集群。文獻[15]在文獻[14]的基礎上提出異步的節點性能監控策略，通過性能監控(Quality Monitor， QM)進程和性能反饋(Quality Reporter， QR)進程異步監控節點的性能數據，有效降低了作業執行的時間開銷，并將該算法部署于200個節點的大規模集群，但該策略監控的性能指標較單一，且未考慮節點間的傳輸開銷。文獻[16]在文獻[15]的基礎上建立數學模型，依據QR和QM收集的性能數據算出每個算子的合理并行度，并根據計算結果進行動態調整，從而在滿足計算延遲約束的前提下有效提高集群的吞吐量，但其數學模型過于復雜，集群在輸入速率急劇上升階段的響應速度無法滿足實際需求。文獻[17]提出基于有狀態數據分片調度策略的數據流系統ChronoStream，通過實施高效的狀態數據管理計劃，使節點在橫向和縱向上均實現可伸縮性，但其分片的調度策略產生了較高的時間開銷。文獻[18]提出一種可擴展的數據流處理系統StreamCloud，通過整合高效的任務調度和負載均衡策略，實現對用戶透明的數據流查詢功能，其思想被用于改進Borealis平臺并取得了很好的效果。文獻[19]根據集群拓撲中關鍵路徑上的性能感知數據，在保證計算實時性的前提下盡可能降低能耗，但未考慮節點內存、網絡傳輸等其他性能指標對集群性能的影響。文獻[20]提出用計算延遲作為綜合評估節點性能的指標，通過實施節點間的動態負載均衡策略降低任務的計算延遲。

針對上述文獻中存在的數據流任務調度策略多關注節點內的計算開銷，而忽略節點間傳輸開銷的問題，本文的主要工作有：

1)通過定義流式計算的有向無環圖(Directed Acyclic Graph， DAG)中每條邊的容量與流量值，將其轉化為流網絡模型，兼顧節點的計算開銷與邊的傳輸開銷。

2)提出容量檢測算法，在計算延遲閾值的約束下檢測每個節點的最高負載，并將其記為對應輸入邊的容量，從而構建流網絡模型。

3)在流網絡模型的基礎上提出最大流算法，在輸入速率上升階段根據流量與容量的關系進行合理的負載分配，在滿足延遲約束的前提下提供盡可能高的吞吐量，實現計算資源的最大化利用。

2 流網絡模型

通過動態調度策略合理分配新增的計算負載，最大化利用計算資源，才能在輸入速率上升階段有效提高集群的吞吐量。如果將數據源的輸入速率作為期望吞吐量(Expected Throughput， ET)，而集群當前時刻實際處理數據的速率為實際吞吐量(Actual Throughput， AT)，則動態調度策略的目的是通過優化節點間的調度和負載分配方式，使集群的實際吞吐量滿足不斷上升的期望吞吐量。最大流算法通過建立流網絡模型，尋找一條從源點到匯點的優化路徑，并沿著優化路徑的方向提高計算負載，從而提高整個集群的實際吞吐量。

2.1 流式計算的結構

在大數據流式計算中，通常將用戶定義功能(User Define Function， UDF)作為一系列算子，待處理的數據元組從源點發出，依次經過每個算子的處理，最終將計算結果在匯點持久化。其中數據源點往往可以有多種多樣的數據產生方式，數據匯點可以是Hadoop分布式文件系統(Hadoop Distributed File System, HDFS)等數據存儲平臺或直接將處理結果反饋給用戶，中間的一系列算子共同實現了用戶定義的業務功能。

在分布式數據流處理系統中，為了提高集群的性能以保證計算的實時性，通常將同一個算子映射到不同的計算節點上，使它們能夠分別同時完成相同的計算任務，從而提高任務的執行效率。如圖1所示，O1、O2、O3是任務中依次處理數據的3個算子，被分別映射到v1、v2等7個計算節點上，算子之間數據傳輸被映射到計算節點間的通信鏈路上，這樣就形成了流式計算的DAG拓撲；但傳統的流式計算模型大多只關注節點內部的計算延遲，而忽略了節點間邊的傳輸延遲。事實上集群往往受限于計算和傳輸共同導致的時間開銷，而難以實現低延遲和高吞吐量兼得，急劇上升的計算負載會導致數據被阻塞而產生更高的延遲，因此，有效的任務調度策略必須兼顧節點內部的計算開銷和節點間的傳輸開銷，并在滿足延遲約束的前提下盡可能提高實際吞吐量。

圖1 流式計算模型

2.2 數據流網絡

通過定義DAG拓撲中每條邊上允許數據傳輸的最大速率為該邊的容量，而實際傳輸的速率為流量，就形成了對應的流網絡模型。

定義1 數據流網絡。如圖2所示，設有向無環圖G=(V,E)，其中V={v1,v2,…,vn}是圖中所有節點的集合，s∈V是流網絡的源點，t∈V是匯點，E={(vi,vj)|i,j∈[1,n],n=|V|}是所有邊的集合，(vi,vj)是從節點vi向vj傳輸數據的邊。其中每條邊(vi,vj)∈E都有c(vi,vj)≥0表示邊(vi,vj)允許數據傳輸速率的最大值，也稱為邊(vi,vj)的容量，而實際從節點vi向vj傳輸數據的速率是邊(vi,vj)的流量，記為f(vi,vj)。

根據定義1可知，對于流網絡中任意一條邊(vi,vj)∈E,都有0≤f(vi,vj)≤c(vi,vj)，即在任意邊上傳輸數據的速率不能超過其容量的限制，這稱為容量限制定律；同時，對于任意的節點vi∈V-{s,t}，其所有的前驅節點記為vj，后繼節點記為vk，則滿足：

(1)

即對于流網絡中任意一個計算節點，受其內部計算開銷的影響，在任意時刻數據流入該節點的速率總是大于或等于數據流出該節點的速率，這稱為流量限制定律。實際上，流網絡中任意邊(vi,vj)的容量值c(vi,vj)的大小都與節點vj及其后繼的數據處理能力有關：節點vj的處理能力越強、局部吞吐量越大，則c(vi,vj)越大；反之c(vi,vj)越小。同時，每條邊的容量大小還與節點間的網絡傳輸速率、計算延遲約束等多種因素有關，而流量f(vi,vj)是在任務運行中的某一時刻，實際從節點vi向vj傳輸數據的速率，是隨著時間不斷變化的。

圖2 數據流網絡圖

定義2 流。設G=(V,E)是一個流網絡，其中s為源點，t為匯點，則G的流是一個實值函數f：V×V→R。其流量的大小為：

(2)

流網絡中一個流的流量是數據從源點流出速率的和也是數據流入匯點的速率的和，是集群實際處理數據的速率，即當前時刻的實際吞吐量，其中流量最大的一個流是G的最大流，記為fmax。

定義3 增進網絡。如圖3所示，設流網絡G=(V,E)，則其對應的增進網絡為Gf=(Vf,Ef)，其中對于所有的節點vi∈V都有vi∈Vf，對于所有的邊(vi,vj)∈E，在增進網絡中對應的容量cf(vi,vj)為：

(3)

其中：E是原網絡中邊的集合；c(vi,vj)是原網絡中邊(vi,vj)的容量；f(vi,vj)是原網絡中邊(vi,vj)的流量。

圖3 增進網絡圖

根據定義3可知，增進網絡主要反映了對應原網絡中流量可能提升的空間，其中存在與原網絡中反向的邊，是因為在優化負載分配的過程中，有可能減少一些邊的流量而增加到另外一些邊上，實現提升整個集群吞吐量的目的，因此，在增進網絡中尋找一條優化路徑就可以按照其方向提高原網絡的流量。

|fp|=cf(p)=min {cf(vi,vj)|(vi,vj)∈P}

(4)

其中cf(vi,vj)是增進網絡中邊(vi,vj)的容量。

優化路徑是提升原網絡流量的一個方案：當期望吞吐量上升時，系統通過在增進網絡中尋找一條優化路徑，并在原網絡中將優化路徑上的邊的流量分別增大|fp|，就得到一條流量為|f|+|fp|的流。通過這樣反復迭代，不斷在增進網絡中尋找新的優化路徑就可以不斷提高集群的實際吞吐量。

定理1 最大流定理。設流網絡G=(V,E)，Gf是其對應的增進網絡，f是流網絡G的一個流，則以下兩個條件是互相等價的：

條件1f是G的最大流，即|f|=|fmax|；

條件2 增進網絡中不存在任何優化路徑。

證明

證畢。

根據定理1可知，流網絡達到最大流當且僅當對應的增進網絡中不存在任何優化路徑，即只要在增進網絡中能找到一條新的優化路徑，就可以沿著優化路徑的方向提升原網絡的流量，這為提出最大流算法提供了模型的支撐。

3 最大流算法

基于流網絡模型及其相關定義，FNDD策略先通過容量檢測算法確定DAG拓撲中每條邊的容量值，將其轉化為流網絡模型。在輸入數據速率上升階段，當期望吞吐量大于集群的實際吞吐量時，首先根據流網絡中每條邊上容量與流量的差值，通過最大流算法計算對應的增進網絡并尋找一條優化路徑，再通過沿著優化路徑的方向提升原網絡的流量，實現在限定的延遲約束下提升實際吞吐量的目標。

3.1 容量檢測算法

只有將流式計算的DAG拓撲轉化為流網絡模型，才能使用最大流算法提高集群的實際吞吐量，因此在限定的延遲約束下確定每條邊的容量大小，對最大流算法的執行效果至關重要。容量過大會導致節點在實際環境中無法及時處理數據，使其在緩存中被滯留而延遲加長，甚至因內存耗盡導致節點崩潰，而容量過小則無法充分利用計算資源。

為了在限定的延遲約束下獲得盡可能高的吞吐量，必須在任務啟動后，首先通過容量檢測算法確定每條邊的容量值，從而為最大流算法的執行建立流網絡模型。算法在限定計算延遲閾值的前提下不斷提高期望吞吐量，當實際的延遲遠小于設定的閾值時，以恒定的步長提高期望吞吐量；當實際的延遲略小于或等于閾值時，將當前的期望吞吐量作為節點對應輸入邊的容量。當所有邊的容量值都確定后，就完成了流網絡模型的構建。容量檢測算法的具體執行過程如算法1所示。

算法1 容量檢測算法。

輸入：集群拓撲G′，延遲約束的閾值θ，期望吞吐量ET。

輸出：數據流網絡G。

1)

foreache∈G.E

2)

e.c← ∞;

/*將DAG中所有邊的容量初始化為無窮大*/

3)

end foreach

4)

varnum← |G.E|;

/*用變量num記錄尚未確定容量值的邊的數目*/

5)

whilenum>0

6)

G.s.start(ET,60);

/*作業開始執行的第1 min，以ET的速率向集群輸入數據*/

7)

foreachv∈G.V

/*依次遍歷流網絡中的每一個節點*/

8)

if avg(v.f-v.d)-θ≤εthen

/*尋找平均計算延遲略小于或等于閾值θ的節點*/

9)

v.pe.c←ET;

/*將當前節點輸入邊的容量設為當前的期望吞吐量*/

10)

num--;

/*待確定容量值的邊的數目減1*/

11)

end if

12)

end foreach

13)

ET←ET+10 000;

/*提升期望吞吐量，準備進入下一次迭代*/

14)

end while

15)

returnG;

如算法1所示，首先將拓撲中所有邊的容量設為無窮大(第1)～3)行)并記錄拓撲中邊的數目(第4)行)，然后以用戶設定的初始ET從源點開始輸入數據(第6)行)，每經過60 s統計一次平均計算延遲并尋找所有延遲略小于或等于閾值θ的節點，將當前的ET作為其對應輸入邊的容量并將未確定容量的邊數減1(第8)～11)行)，最后判斷如果拓撲中還有未確定容量值的邊，則提高ET的大小并進入下一次迭代(第13)行)，直到所有邊都確定容量為止。這樣就將流式計算的DAG拓撲轉化為對應的流網絡模型，同時保證當每條邊都滿足容量限制定律時，計算延遲應當不超過設定的延遲閾值，為最大流算法提供了模型的支撐。

3.2 最大流算法

根據流網絡及其相關定義，在容量檢測算法確定每條邊的容量大小后，當期望吞吐量大于實際吞吐量時，就可以通過最大流算法增加一些邊的流量以提高整個集群的實際吞吐量：首先根據定義3計算流網絡對應的增進網絡，然后用圖的廣度優先搜索算法在增進網絡中尋找一條優化路徑P，再根據定義4計算優化路徑所對應的增量|fp|，最后在原網絡中沿著優化路徑的方向提高每條邊的流量，并將提升后的流量記為：

(5)

則整個流網絡的流量大小提升至|f|+|fp|，其中f(vi,vj)為原網絡中邊(vi,vj)的流量。

根據增進網絡、優化路徑和流網絡中每條邊上流量與容量的大小關系，當期望吞吐量大于集群的實際吞吐量時調用最大流算法提升集群的吞吐量。最大流算法的具體執行過程如算法2所示。

算法2 最大流算法。

輸入：流網絡G;期望吞吐量ET。

輸出：提升后的流量|f|。

1)

Gf.V←G.V;

/*根據定義3，原網絡的節點集合就是

增進網絡的節點集合*/

2)

foreach (vi,vj)∈G.E

3)

cf(vi,vj) ← (vi,vj).c-(vi,vj).f;

4)

cf(vi,vj) ← (vi,vj).f;

5)

end foreach

/*根據定義3計算增進網絡中對應邊的容量*/

6)

P← BFS(Gf,s,t);

/*通過廣度優先搜索在增進網絡中

尋找一條從源點s到匯點t的優化路徑*/

7)

whileET>|G.f| andP!=?

/*當期望吞吐量大于流量且

增進網絡中存在優化路徑時，進入提升網絡流量的迭代過程*/

8)

|fp| ← min{cf(vi,vj)|(vi,vj)∈P};

/*計算優化路徑對應的增量*/

9)

foreach edge(vi,vj)∈P

10)

if (vi,vj)∈G.E

11)

(vi,vj).f← (vi,vj).f+|fp|;

12)

else if (vj,vi)∈G.E

13)

(vj,vi).f← (vj,vi).f-|fp|;

14)

end if

/*根據式(5)，沿著優化路徑的

方向提升原網絡的流量*/

15)

end foreach

16)

|G.f| ← |G.f|+|fp|;

/*記錄新的流網絡的

流量大小*/

17)

P← BFS(Gf,s,t);

/*尋找新的優化路徑并

進入下一次迭代*/

18)

end while

19)

return |G.f|;

算法先根據流網絡構建對應的增進網絡(第1)～5)行)并在增進網絡中用廣度優先搜索算法尋找一條優化路徑(第6)行)，如果存在優化路徑就進入對原網絡的迭代優化過程：首先根據定義4計算優化路徑對應的增量(第8)行)，再根據式(5)提高原網絡中對應邊的流量(第9)～15)行)，最后記錄新的網絡流量并尋找一條優化路徑進入下一次迭代。

根據定理1可知，只要增進網絡中存在優化路徑就意味著原網絡的流量仍有提升的空間，沿著優化路徑的方向就可以提升集群的吞吐量，使實際吞吐量不斷滿足期望吞吐量的要求。直到增進網絡中不存在任何優化路徑時，集群中所有節點都處于滿負荷工作狀態，此時計算資源得到最大化利用。

3.3 參數影響與代價評估

閾值θ是FNDD策略中唯一的參數，是由用戶定義的作業中允許每個數據元組的最大計算延遲，取值過小會導致集群能承受的負載過低，而取值過大則無法滿足作業的實時性要求。實際上θ的取值與以下三個因素有關：其一，與作業本身的復雜度有關，作業的復雜度越高則θ的取值應越大，反之可以設定較小的θ值；其二，與實際應用中對服務質量的要求有關，用戶對計算的實時性要求越高θ的取值應該越小；其三，與集群的實際規模和性能有關，集群的節點數越多、計算能力越強則計算延遲越低，θ的取值也可相應減小。這三個因素都是在算法設計和實現過程中無法掌握的，因此由用戶根據應用中作業和集群的實際情況設定，4.2節通過實驗得出在每種作業類型下推薦的參數值范圍，供用戶參考。

在算法的復雜度方面，容量檢測算法的時間復雜度為T(n)=O(|V|×|E|)，其中|V|和|E|分別為流網絡中節點和邊的數目，目前Flink平臺在實際應用中的最大集群規模約1 500個節點[21]，節點間邊的數目與實際應用中集群的拓撲結構和作業的部署模型有關，且|E|≤|V|×k/2，其中k與任務的并行度和集群的拓撲結構有關，當k=10時，|E|≤1 500×10/2=7 500，因此容量檢測算法的時間開銷在合理可接受的范圍內。另外算法收斂的速度還與期望吞吐量遞增的步長有關，設定合適的步長能夠使整個流網絡更快地趨于穩定。最大流算法的執行效率與在增進網絡中尋找優化路徑的算法密切相關，使用廣度優先搜索算法選擇優化路徑的時間復雜度為T(n)=O(|V|+|E|)=O(|E|)。最大流算法的執行還與提升的流量有關：設最大流為|fmax|，則如果每次迭代增加1 tuple/s時算法達到最高時間復雜度為T(n)=O(|E|×|fmax-f|)，其中|f|是集群當前的流量，這在實際應用中是不太可能出現的。由于流式計算集群的節點以及節點間通信鏈路的數目都不是很高，因此整個FNDD策略的時間復雜度是可接受的。在空間復雜度上，流網絡模型只在DAG拓撲的基礎上改變了每條邊的權值而沒有帶來新的空間開銷，而增進網絡與流網絡的空間復雜度是相等的，同時實驗驗證了FNDD策略對集群性能的優化遠大于算法本身的開銷，因此算法在時間和空間復雜度上都是可行的。

4 實驗結果與分析

Apache Flink是目前應用中最重要的數據流處理平臺之一，承擔著許多企業的實時計算任務。為了使FNDD策略能夠更好地在實踐中得到應用，在Flink平臺中實現了最大流和容量檢測算法，并針對不同作業類型的基準測試選定了能夠使算法達到最優效果的參數值，最后在相同環境下分別從吞吐量、計算延遲以及內存占用率三個維度將FNDD策略與原系統的調度策略形成對比，驗證了算法的優化效果。

4.1 實驗環境

實驗搭建的集群由15臺普通物理PC機組成，分別由Kafka作為數據源點，根據實驗設置以不同的速率向集群輸入數據，用TaskManager節點構建整個計算拓撲，將計算結果保存在HDFS中并統計相關性能指標，以Zookeeper作為集群的同步協調節點負責分布式節點間的信息同步。集群中所有節點都連接在一個獨立的專用網絡中，與公共網絡隔離，不產生任何非必要的額外傳輸開銷。具體的節點分布情況如表1所示。

表1 集群節點分布信息

集群中所有節點采用相同的軟硬件配置環境，配置參數如表2所示。每個TaskManager只開啟一個TaskSlot，即參數taskmanager.numberOfTaskSlots=1，因此作業的并行度最大開啟到10，即parallelism.default=10。這樣可以充分利用計算資源，并驗證FNDD策略在不同計算節點之間進行作業調度的優化效果，避免在同一個節點內的不同進程間進行負載分配。

表2 節點配置參數

實驗分別執行了WordCount、TwitterSentiment和TeraSort三個標準的基準測試，首先通過參數調整實驗分別確定了在每種類型的作業中，能夠使算法達到最優效果的參數θ的取值，再分別將FNDD策略與Flink系統原生的調度策略進行對比，驗證了算法的優化效果。

4.2 參數調整實驗

為了確定參數θ的取值范圍，使集群達到最高實際吞吐量，即FNDD策略達到最好的優化效果，首先在不同的作業類型下開展參數調整實驗。實驗選取的3個基準測試分別代表了流式計算3種不同類型的作業：WordCount用于統計英文單詞出現的頻次，其計算復雜度低且對內存的占用率較低，但對CPU資源的占用率較高；TwitterSentiment是Twitter公司開發的對用戶發布的推文進行實時情感分析的作業，其計算相對復雜且對CPU和內存資源的占用率都比較高；TeraSort是對大規模數據進行分布式排序的作業，計算復雜度最高，作業執行過程中產生大量狀態數據會占用內存資源且節點間有頻繁的數據交互。

根據對原系統的采樣結果可知：3個作業執行中的計算延遲大多分布在0.1 ms～0.2 ms，最高實際吞吐量不超過90 000 tuple/s，因此，為了選取參數θ更精確的取值以獲得最高的實際吞吐量，實驗將期望吞吐量設為95 000 tuple/s，θ在0.1 ms～0.2 ms以0.01為步長依次取值，得到如圖4所示的實驗結果。

圖4 不同參數的吞吐量對比

根據容量檢測算法的核心思想，當算法在不同的參數取值下得到非常相近的吞吐量時，實驗總是選擇盡可能小的θ取值，通過限定較低的計算延遲約束來提高計算的實時性。根據圖4可知，WordCount作業在θ取0.13 ms～0.20 ms時都能達到最高吞吐量89 500 tuple/s，因此選擇最小值θ=0.13 ms。同理可得，TwitterSentiment作業能達到最高吞吐量69 700 tuple/s的最小θ值為0.15 ms，TeraSort作業能達到最高吞吐量49 000 tuple/s的最小θ值為0.17 ms。

為了進一步驗證參數θ的取值，在獲得高吞吐量的同時盡可能降低延遲，實驗檢測在不同參數值下的計算延遲并進行對比。根據圖4可知，計算復雜度最高的TeraSort作業的最高吞吐量平均可達5 000 tuple/s。為了避免過高的輸入速率造成數據阻塞而影響計算延遲的檢測結果，實驗將3個作業的期望吞吐量固定在50 000 tuple/s，分別在不同的參數下執行作業并統計實際的平均計算延遲，得到如圖5所示的實驗結果。這與吞吐量對比實驗中得到的結果是基本一致的，WordCount作業在θ=0.13 ms時達到最低的延遲，TwitterSentiment作業在θ=0.15 ms時達到最低的延遲，TeraSort作業在θ=0.17 ms時達到最低的延遲。

綜上所述，3種類型作業的參數取值都在0.1 ms～0.2 ms，根據圖4和圖5可知，當計算比較簡單時其延遲相對較低，則參數取值一般不超過0.15 ms，當計算任務相對復雜時θ的取值應有所增大，一般在0.15 ms～0.17 ms。而排序類作業計算復雜且內存占用率高，因此參數θ的取值一般在0.17 ms以上。通過分析在不同作業類型下的實驗結果，確定了參數θ的合理取值范圍，能夠使集群達到最高實際吞吐量，FNDD策略實現較好的優化效果。

4.3 對比實驗與分析

根據參數調整實驗得到的實驗結果，分別確定了參數θ的合理取值范圍，因此對比實驗使用該取值分別執行WordCoud、TwitterSentiment和TeraSort作業，以驗證FNDD策略的優化效果。

圖5 不同參數的計算延遲對比

其中WordCount作業的計算本身并不復雜，但其作業執行過程中對節點的CPU占用率較高，是常用的測試集群性能的標準基準測試。由圖4可知，WordCount作業的最高吞吐量約90 000 tuple/s。為了驗證FNDD策略在輸入速率上升階段的優化效果，實驗將初始的期望吞吐量設為40 000 tuple/s，每經過1 min將期望吞吐量提高10 000 tuple/s，直至期望吞吐量達到90 000 tuple/s后持續輸入3 min，之后期望吞吐量逐步下降，并從吞吐量和計算延遲兩個維度將FNDD策略與原系統調度策略的性能形成對比。

如圖6所示，隨著期望吞吐量的逐步上升，Flink原系統在約68 000 tuple/s時達到其吞吐量的瓶頸，當期望吞吐量繼續上升時有數據元組被阻塞而延遲加長，在未開啟檢查點機制時甚至出現數據丟棄的現象。通過使用FNDD策略，當期望吞吐量不斷上升時，算法根據優化路徑的方向合理分配新增的計算負載，使集群的實際吞吐量從68 000 tuple/s提高至88 000 tuple/s，平均提高了29.41%，基本滿足期望吞吐量的要求。另外通過實驗發現參數θ取0.13 ms或0.15 ms時都能取得比較好的優化效果，但當θ=0.15 ms時在期望吞吐量上升階段的優化效果更顯著，最終兩種情況都穩定于幾乎相同的吞吐量值，但在計算延遲上有比較明顯的區別。

圖6 WordCount吞吐量對比

圖7為匯點每接收到10 000 tuple時記錄一個延遲時間并持續12 min得到的實驗結果：在原系統中由于部分節點無法及時處理數據，導致部分元組被阻塞而計算延遲加長，而經過FNDD策略優化后集群的計算延遲有較明顯的下降。當θ=0.13 ms時雖然在輸入速率上升階段的實際吞吐量上升較慢，但比θ=0.15 ms時的計算延遲更低。

TwitterSentiment作業相對于WordCount的計算更復雜，在相同環境下達到的實際吞吐量較低，因此根據參數調整實驗的分析結果，實驗設置的期望吞吐量從20 000 tuple/s遞增到70 000 tuple/s，參數θ的取值分別為0.15 ms和0.17 ms。

如圖8所示，由于作業本身計算復雜度高，實驗設置的期望吞吐量最高達70 000 tuple/s，但原系統的實際吞吐量在約59 000 tuple/s時達到瓶頸。經FNDD策略的優化將實際吞吐量平均提高到68 500 tuple/s，較原系統平均提高了16.12%，受資源總量和作業復雜度的限制，其優化效果不是非常明顯，但已有效提高了實際吞吐量。

圖7 WordCount延遲對比

圖8 TwitterSentiment吞吐量對比

如圖9所示，TwitterSentiment作業的計算延遲本身較高，其優化效果也相對明顯：原系統在期望吞吐量上升時的延遲上升比較顯著，通過算法優化將每1萬條數據的計算延遲最多降低了416 ms，提高了計算的實時性；但兩種參數取值下的延遲相差比較明顯，當θ=0.17 ms時能夠獲得比較高的吞吐量，但其計算延遲也明顯較高。

圖9 TwitterSentiment延遲對比

TeraSort作業的計算復雜度和內存占用率最高，且計算過程中節點間有頻繁的數據交互，根據參數調整實驗的分析結果，將參數θ設為0.17 ms和0.19 ms，分別從吞吐量和內存占用率兩個維度將FNDD策略與原系統的調度策略形成對比。

如圖10所示，輸入的最高期望吞吐量為50 000 tuple/s，而原系統能達到的最高實際吞吐量只有33 000 tuple/s，且計算延遲較高。通過FNDD策略的優化，集群的實際吞吐量最高可達到49 000 tuple/s，較原系統的實際吞吐量平均提高了38.29%，最大化利用了現有的計算資源且基本滿足了期望吞吐量的要求，其中當θ=0.19 ms時的吞吐量能夠穩步上升，集群的穩定性較高；但算法的優化是一個逐步提高吞吐量的過程，因此期望吞吐量達到40 000 tuple/s時保持穩定1 min，算法的執行過程有一定的時間開銷，隨著算法的執行集群的吞吐量進一步上升。

圖10 TeraSort吞吐量對比

為了進一步驗證FNDD策略對高復雜度作業的優化效果，實驗在TeraSort作業執行過程中實時監控節點的內存占用率，通過定點采樣得到如圖11所示的實驗結果：當期望吞吐量上升時，原系統將單位時間內新增的數據元組分配給一部分節點，導致其負載過高而內存占用率急劇上升，而另外一部分節點的資源未得到充分利用，導致部分節點無法及時處理數據而延遲加長。通過使用FNDD策略，使優化后集群被采樣節點的內存占用率都有一定程度的上升且基本趨于穩定，每個有剩余資源的節點都分擔了新增的計算負載，通過避免數據阻塞降低了計算延遲，實現節點間的負載均衡的同時穩步提高吞吐量。

圖11 TeraSort內存占用率對比

綜上所述，實驗表明FNDD策略在期望吞吐量上升階段對集群的性能有一定的優化作用，通過檢測每條邊上容量與流量的差值，對新增的數據元組進行更合理的負載分配。在不同的作業類型下，該策略對原系統吞吐量的優化效果并不相同，但其平均優化比均高于16.12%。算法通過最大化利用集群的計算資源，在滿足計算延遲約束的前提下有效提高了集群的實際吞吐量。

5 結語

由于數據源的多樣性和輸入速率的急劇變化給流式計算集群造成極大的負載壓力，進而影響了計算的實時性和準確性，因此，本文提出基于流網絡模型的動態調度策略，關注每個計算節點和傳輸鏈路的性能，在輸入速率急劇上升時根據每條邊上容量與流量的關系進行合理的負載分配，有效提高了集群的吞吐量；但FNDD策略關注集群輸入速率急劇上升階段的性能優化，這一階段節點的計算和響應能力處于基本穩定狀態，因此策略在作業開始時確定鏈路的容量大小。在任務執行的其他階段，特別是在輸入速率出現劇烈波動時，根據作業的執行情況動態調整容量的大小，能最大化利用集群的計算資源，因此，為了使FNDD策略能夠適用于任務執行的各個階段，下一步研究將重點關注容量的動態變化問題，根據作業執行情況和節點的剩余資源動態調整鏈路容量的大小，從而在任務執行的其他階段取得更好的優化效果。