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

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

0 引言

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

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

1 相關(guān)工作

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

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

針對上述文獻(xiàn)中存在的數(shù)據(jù)流任務(wù)調(diào)度策略多關(guān)注節(jié)點內(nèi)的計算開銷，而忽略節(jié)點間傳輸開銷的問題，本文的主要工作有：

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

2)提出容量檢測算法，在計算延遲閾值的約束下檢測每個節(jié)點的最高負(fù)載，并將其記為對應(yīng)輸入邊的容量，從而構(gòu)建流網(wǎng)絡(luò)模型。

3)在流網(wǎng)絡(luò)模型的基礎(chǔ)上提出最大流算法，在輸入速率上升階段根據(jù)流量與容量的關(guān)系進(jìn)行合理的負(fù)載分配，在滿足延遲約束的前提下提供盡可能高的吞吐量，實現(xiàn)計算資源的最大化利用。

2 流網(wǎng)絡(luò)模型

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

2.1 流式計算的結(jié)構(gòu)

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

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

圖1 流式計算模型

2.2 數(shù)據(jù)流網(wǎng)絡(luò)

通過定義DAG拓?fù)渲忻織l邊上允許數(shù)據(jù)傳輸?shù)淖畲笏俾蕿樵撨叺娜萘?，而實際傳輸?shù)乃俾蕿榱髁?，就形成了對?yīng)的流網(wǎng)絡(luò)模型。

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

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

(1)

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

圖2 數(shù)據(jù)流網(wǎng)絡(luò)圖

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

(2)

流網(wǎng)絡(luò)中一個流的流量是數(shù)據(jù)從源點流出速率的和也是數(shù)據(jù)流入?yún)R點的速率的和，是集群實際處理數(shù)據(jù)的速率，即當(dāng)前時刻的實際吞吐量，其中流量最大的一個流是G的最大流，記為fmax。

定義3 增進(jìn)網(wǎng)絡(luò)。如圖3所示，設(shè)流網(wǎng)絡(luò)G=(V,E)，則其對應(yīng)的增進(jìn)網(wǎng)絡(luò)為Gf=(Vf,Ef)，其中對于所有的節(jié)點vi∈V都有vi∈Vf，對于所有的邊(vi,vj)∈E，在增進(jìn)網(wǎng)絡(luò)中對應(yīng)的容量cf(vi,vj)為：

(3)

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

圖3 增進(jìn)網(wǎng)絡(luò)圖

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

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

(4)

其中cf(vi,vj)是增進(jìn)網(wǎng)絡(luò)中邊(vi,vj)的容量。

優(yōu)化路徑是提升原網(wǎng)絡(luò)流量的一個方案：當(dāng)期望吞吐量上升時，系統(tǒng)通過在增進(jìn)網(wǎng)絡(luò)中尋找一條優(yōu)化路徑，并在原網(wǎng)絡(luò)中將優(yōu)化路徑上的邊的流量分別增大|fp|，就得到一條流量為|f|+|fp|的流。通過這樣反復(fù)迭代，不斷在增進(jìn)網(wǎng)絡(luò)中尋找新的優(yōu)化路徑就可以不斷提高集群的實際吞吐量。

定理1 最大流定理。設(shè)流網(wǎng)絡(luò)G=(V,E)，Gf是其對應(yīng)的增進(jìn)網(wǎng)絡(luò)，f是流網(wǎng)絡(luò)G的一個流，則以下兩個條件是互相等價的：

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

條件2 增進(jìn)網(wǎng)絡(luò)中不存在任何優(yōu)化路徑。

證明

證畢。

根據(jù)定理1可知，流網(wǎng)絡(luò)達(dá)到最大流當(dāng)且僅當(dāng)對應(yīng)的增進(jìn)網(wǎng)絡(luò)中不存在任何優(yōu)化路徑，即只要在增進(jìn)網(wǎng)絡(luò)中能找到一條新的優(yōu)化路徑，就可以沿著優(yōu)化路徑的方向提升原網(wǎng)絡(luò)的流量，這為提出最大流算法提供了模型的支撐。

3 最大流算法

基于流網(wǎng)絡(luò)模型及其相關(guān)定義，F(xiàn)NDD策略先通過容量檢測算法確定DAG拓?fù)渲忻織l邊的容量值，將其轉(zhuǎn)化為流網(wǎng)絡(luò)模型。在輸入數(shù)據(jù)速率上升階段，當(dāng)期望吞吐量大于集群的實際吞吐量時，首先根據(jù)流網(wǎng)絡(luò)中每條邊上容量與流量的差值，通過最大流算法計算對應(yīng)的增進(jìn)網(wǎng)絡(luò)并尋找一條優(yōu)化路徑，再通過沿著優(yōu)化路徑的方向提升原網(wǎng)絡(luò)的流量，實現(xiàn)在限定的延遲約束下提升實際吞吐量的目標(biāo)。

3.1 容量檢測算法

只有將流式計算的DAG拓?fù)滢D(zhuǎn)化為流網(wǎng)絡(luò)模型，才能使用最大流算法提高集群的實際吞吐量，因此在限定的延遲約束下確定每條邊的容量大小，對最大流算法的執(zhí)行效果至關(guān)重要。容量過大會導(dǎo)致節(jié)點在實際環(huán)境中無法及時處理數(shù)據(jù)，使其在緩存中被滯留而延遲加長，甚至因內(nèi)存耗盡導(dǎo)致節(jié)點崩潰，而容量過小則無法充分利用計算資源。

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

算法1 容量檢測算法。

輸入：集群拓?fù)銰′，延遲約束的閾值θ，期望吞吐量ET。

輸出：數(shù)據(jù)流網(wǎng)絡(luò)G。

1)

foreache∈G.E

2)

e.c← ∞;

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

3)

end foreach

4)

varnum← |G.E|;

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

5)

whilenum>0

6)

G.s.start(ET,60);

/*作業(yè)開始執(zhí)行的第1 min，以ET的速率向集群輸入數(shù)據(jù)*/

7)

foreachv∈G.V

/*依次遍歷流網(wǎng)絡(luò)中的每一個節(jié)點*/

8)

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

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

9)

v.pe.c←ET;

/*將當(dāng)前節(jié)點輸入邊的容量設(shè)為當(dāng)前的期望吞吐量*/

10)

num--;

/*待確定容量值的邊的數(shù)目減1*/

11)

end if

12)

end foreach

13)

ET←ET+10 000;

/*提升期望吞吐量，準(zhǔn)備進(jìn)入下一次迭代*/

14)

end while

15)

returnG;

如算法1所示，首先將拓?fù)渲兴羞叺娜萘吭O(shè)為無窮大(第1)～3)行)并記錄拓?fù)渲羞叺臄?shù)目(第4)行)，然后以用戶設(shè)定的初始ET從源點開始輸入數(shù)據(jù)(第6)行)，每經(jīng)過60 s統(tǒng)計一次平均計算延遲并尋找所有延遲略小于或等于閾值θ的節(jié)點，將當(dāng)前的ET作為其對應(yīng)輸入邊的容量并將未確定容量的邊數(shù)減1(第8)～11)行)，最后判斷如果拓?fù)渲羞€有未確定容量值的邊，則提高ET的大小并進(jìn)入下一次迭代(第13)行)，直到所有邊都確定容量為止。這樣就將流式計算的DAG拓?fù)滢D(zhuǎn)化為對應(yīng)的流網(wǎng)絡(luò)模型，同時保證當(dāng)每條邊都滿足容量限制定律時，計算延遲應(yīng)當(dāng)不超過設(shè)定的延遲閾值，為最大流算法提供了模型的支撐。

3.2 最大流算法

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

(5)

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

根據(jù)增進(jìn)網(wǎng)絡(luò)、優(yōu)化路徑和流網(wǎng)絡(luò)中每條邊上流量與容量的大小關(guān)系，當(dāng)期望吞吐量大于集群的實際吞吐量時調(diào)用最大流算法提升集群的吞吐量。最大流算法的具體執(zhí)行過程如算法2所示。

算法2 最大流算法。

輸入：流網(wǎng)絡(luò)G;期望吞吐量ET。

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

1)

Gf.V←G.V;

/*根據(jù)定義3，原網(wǎng)絡(luò)的節(jié)點集合就是

增進(jìn)網(wǎng)絡(luò)的節(jié)點集合*/

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

/*根據(jù)定義3計算增進(jìn)網(wǎng)絡(luò)中對應(yīng)邊的容量*/

6)

P← BFS(Gf,s,t);

/*通過廣度優(yōu)先搜索在增進(jìn)網(wǎng)絡(luò)中

尋找一條從源點s到匯點t的優(yōu)化路徑*/

7)

whileET>|G.f| andP!=?

/*當(dāng)期望吞吐量大于流量且

增進(jìn)網(wǎng)絡(luò)中存在優(yōu)化路徑時，進(jìn)入提升網(wǎng)絡(luò)流量的迭代過程*/

8)

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

/*計算優(yōu)化路徑對應(yīng)的增量*/

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

/*根據(jù)式(5)，沿著優(yōu)化路徑的

方向提升原網(wǎng)絡(luò)的流量*/

15)

end foreach

16)

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

/*記錄新的流網(wǎng)絡(luò)的

流量大小*/

17)

P← BFS(Gf,s,t);

/*尋找新的優(yōu)化路徑并

進(jìn)入下一次迭代*/

18)

end while

19)

return |G.f|;

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

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

3.3 參數(shù)影響與代價評估

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

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

4 實驗結(jié)果與分析

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

4.1 實驗環(huán)境

實驗搭建的集群由15臺普通物理PC機(jī)組成，分別由Kafka作為數(shù)據(jù)源點，根據(jù)實驗設(shè)置以不同的速率向集群輸入數(shù)據(jù)，用TaskManager節(jié)點構(gòu)建整個計算拓?fù)洌瑢⒂嬎憬Y(jié)果保存在HDFS中并統(tǒng)計相關(guān)性能指標(biāo)，以Zookeeper作為集群的同步協(xié)調(diào)節(jié)點負(fù)責(zé)分布式節(jié)點間的信息同步。集群中所有節(jié)點都連接在一個獨立的專用網(wǎng)絡(luò)中，與公共網(wǎng)絡(luò)隔離，不產(chǎn)生任何非必要的額外傳輸開銷。具體的節(jié)點分布情況如表1所示。

表1 集群節(jié)點分布信息

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

表2 節(jié)點配置參數(shù)

實驗分別執(zhí)行了WordCount、TwitterSentiment和TeraSort三個標(biāo)準(zhǔn)的基準(zhǔn)測試，首先通過參數(shù)調(diào)整實驗分別確定了在每種類型的作業(yè)中，能夠使算法達(dá)到最優(yōu)效果的參數(shù)θ的取值，再分別將FNDD策略與Flink系統(tǒng)原生的調(diào)度策略進(jìn)行對比，驗證了算法的優(yōu)化效果。

4.2 參數(shù)調(diào)整實驗

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

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

圖4 不同參數(shù)的吞吐量對比

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

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

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

4.3 對比實驗與分析

根據(jù)參數(shù)調(diào)整實驗得到的實驗結(jié)果，分別確定了參數(shù)θ的合理取值范圍，因此對比實驗使用該取值分別執(zhí)行WordCoud、TwitterSentiment和TeraSort作業(yè)，以驗證FNDD策略的優(yōu)化效果。

圖5 不同參數(shù)的計算延遲對比

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

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

圖6 WordCount吞吐量對比

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

TwitterSentiment作業(yè)相對于WordCount的計算更復(fù)雜，在相同環(huán)境下達(dá)到的實際吞吐量較低，因此根據(jù)參數(shù)調(diào)整實驗的分析結(jié)果，實驗設(shè)置的期望吞吐量從20 000 tuple/s遞增到70 000 tuple/s，參數(shù)θ的取值分別為0.15 ms和0.17 ms。

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

圖7 WordCount延遲對比

圖8 TwitterSentiment吞吐量對比

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

圖9 TwitterSentiment延遲對比

TeraSort作業(yè)的計算復(fù)雜度和內(nèi)存占用率最高，且計算過程中節(jié)點間有頻繁的數(shù)據(jù)交互，根據(jù)參數(shù)調(diào)整實驗的分析結(jié)果，將參數(shù)θ設(shè)為0.17 ms和0.19 ms，分別從吞吐量和內(nèi)存占用率兩個維度將FNDD策略與原系統(tǒng)的調(diào)度策略形成對比。

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

圖10 TeraSort吞吐量對比

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

圖11 TeraSort內(nèi)存占用率對比

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

5 結(jié)語

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