基于Flink的任務(wù)調(diào)度策略

何貞貞，于 炯,, 李梓楊，國(guó)冰磊

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

0 引 言

隨著云計(jì)算、大數(shù)據(jù)、物聯(lián)網(wǎng)、人工智能等信息技術(shù)的快速發(fā)展和傳統(tǒng)產(chǎn)業(yè)的數(shù)字化轉(zhuǎn)型，預(yù)計(jì)到2020年我國(guó)數(shù)據(jù)總量將達(dá)到8060 EB，占據(jù)全球數(shù)據(jù)總量的18%[1-4]。在這種爆炸式數(shù)據(jù)量增長(zhǎng)情況下，其規(guī)?？梢赃_(dá)到PB級(jí)別，產(chǎn)生速度可以達(dá)到GB/s級(jí)別[5]，且數(shù)據(jù)的時(shí)效性很強(qiáng)。對(duì)于這些連續(xù)不斷的數(shù)據(jù)，目前的大多數(shù)解決方案卻不是把實(shí)時(shí)流數(shù)據(jù)當(dāng)作流來處理，忽略了其數(shù)據(jù)產(chǎn)生的連續(xù)性和及時(shí)性。為了滿足這種實(shí)時(shí)性要求，流式計(jì)算[6]環(huán)境應(yīng)具備較高的響應(yīng)能力和較低的計(jì)算延遲，同時(shí)要求計(jì)算結(jié)果的準(zhǔn)確性和可靠性。Apache Flink[7-13]是目前產(chǎn)業(yè)界應(yīng)用最廣泛的新興流式計(jì)算平臺(tái)之一，在美團(tuán)、淘寶的實(shí)時(shí)業(yè)務(wù)中已有一定應(yīng)用。

在Apache Flink中，任務(wù)是執(zhí)行算子的并行化實(shí)例的基本單元，每個(gè)任務(wù)在一個(gè)進(jìn)程中執(zhí)行，且Flink中的任務(wù)執(zhí)行圖具有層次分明的拓?fù)浣Y(jié)構(gòu)。因此，為解決Flink計(jì)算平臺(tái)拓?fù)渲幸蚋麝P(guān)鍵節(jié)點(diǎn)上任務(wù)間不同類型通信所導(dǎo)致的通信開銷較大問題，本文提出基于一種Flink環(huán)境下的任務(wù)調(diào)度策略(task scheduling strategy in Flink，TSS-Flink)，該策略是通過動(dòng)態(tài)的調(diào)整關(guān)鍵路徑上各節(jié)點(diǎn)實(shí)例的任務(wù)分配，在保證關(guān)鍵路徑節(jié)點(diǎn)負(fù)載差異較小的同時(shí)降低通信開銷，從而降低關(guān)鍵路徑的響應(yīng)時(shí)間，提高系統(tǒng)性能。同時(shí)，當(dāng)數(shù)據(jù)流壓力發(fā)生變化后，只需要調(diào)整關(guān)鍵路徑上部分節(jié)點(diǎn)任務(wù)，不會(huì)引入更多的任務(wù)調(diào)度開銷。經(jīng)實(shí)驗(yàn)驗(yàn)證得出，該策略對(duì)不同類型的benchmark作業(yè)都有較為明顯的優(yōu)化效果，在保證系統(tǒng)穩(wěn)定性的同時(shí)使計(jì)算延遲平均降低了13.09%。

1 相關(guān)工作

在流式計(jì)算中，通常使用有向無環(huán)圖(directed acyclic graph，DAG)來描述大數(shù)據(jù)流的計(jì)算過程。在拓?fù)渲?，任何情況下都會(huì)存在一條關(guān)鍵路徑，且該關(guān)鍵路徑的計(jì)算延遲決定了整個(gè)任務(wù)拓?fù)涞臅r(shí)延。為了保證資源負(fù)載差異較小的同時(shí)提高計(jì)算的實(shí)時(shí)性，必須提出一種一方面要有效適應(yīng)數(shù)據(jù)流、資源等動(dòng)態(tài)變化時(shí)所帶來的負(fù)載差異較大問題，另一方面也要避免因任務(wù)與任務(wù)之間不同類型的通信帶來巨大的開銷所造成的實(shí)時(shí)性問題。在保證各節(jié)點(diǎn)計(jì)算資源充分利用的前提下，最大程度降低計(jì)算延遲，提高計(jì)算實(shí)時(shí)性。

針對(duì)負(fù)載不均和任務(wù)間通信開銷較大問題，已有大量的學(xué)者開展相關(guān)研究。文獻(xiàn)[14]提出一種Storm環(huán)境下基于權(quán)重的任務(wù)調(diào)度算法，針對(duì)各個(gè)任務(wù)的CPU負(fù)載占用情況以及任務(wù)間的數(shù)據(jù)流大小，分別確定點(diǎn)權(quán)和邊權(quán)，并利用最大化邊權(quán)增益的思想，降低網(wǎng)絡(luò)傳輸開銷。但該算法只考慮了CPU負(fù)載對(duì)系統(tǒng)性能的影響，并未考慮內(nèi)存資源和網(wǎng)絡(luò)帶寬資源對(duì)節(jié)點(diǎn)負(fù)載均衡的影響。文獻(xiàn)[15]提出一種實(shí)時(shí)和高效的資源調(diào)度模型Re-Stream，在大數(shù)據(jù)流式計(jì)算環(huán)境下實(shí)現(xiàn)高能效和低延遲，并結(jié)合拓?fù)鋱?zhí)行關(guān)鍵路徑，提出對(duì)工作節(jié)點(diǎn)的內(nèi)存電壓調(diào)控節(jié)能策略，該策略主要針對(duì)關(guān)鍵路徑和非關(guān)鍵路徑上的內(nèi)存電壓節(jié)能，并未考慮到內(nèi)存資源和網(wǎng)絡(luò)帶寬對(duì)系統(tǒng)的整體性能影響。文獻(xiàn)[16,17]提出高效的資源調(diào)度算法和優(yōu)化框架，雖然解決了流式計(jì)算框架下的任務(wù)調(diào)度問題，但無法直接移植到Flink平臺(tái)。文獻(xiàn)[18]提出一種Storm框架下資源感知的任務(wù)調(diào)度策略，通過最大化資源利用率的同時(shí)最小化網(wǎng)絡(luò)延遲提高吞吐量。文獻(xiàn)[19]提出用計(jì)算延遲作為評(píng)估節(jié)點(diǎn)間負(fù)載的指標(biāo)，通過降低任務(wù)的計(jì)算延遲達(dá)到負(fù)載均衡的效果，但該策略同時(shí)也帶來了較大的遷移開銷，且資源評(píng)估不全面。文獻(xiàn)[20]提出一種基于流網(wǎng)絡(luò)的流式計(jì)算動(dòng)態(tài)任務(wù)調(diào)度策略，通過定義有向無環(huán)圖中每條邊的容量和流量將其轉(zhuǎn)化為流網(wǎng)絡(luò)模型，計(jì)算對(duì)應(yīng)的增進(jìn)網(wǎng)絡(luò)和優(yōu)化路徑來提升集群吞吐量從而提升性能。

現(xiàn)有的研究多關(guān)注于節(jié)點(diǎn)內(nèi)的計(jì)算開銷，不僅忽略了節(jié)點(diǎn)間的不同類型通信方式帶來的傳輸開銷，而且忽略了拓?fù)潢P(guān)鍵路徑對(duì)集群性能的重要影響。且大多數(shù)的已有研究并不適用于Apache Flink平臺(tái)。針對(duì)上述問題，本文的主要工作有：

(1)通過定義流式計(jì)算中的有向無環(huán)圖(DAG)，將數(shù)據(jù)流大小作為邊權(quán)將拓?fù)滢D(zhuǎn)化為AOE-網(wǎng)(activity on edge network)，確定拓?fù)渲嘘P(guān)鍵路徑；

(2)提出負(fù)載均衡模型，主要針對(duì)關(guān)鍵路徑上負(fù)載較高的節(jié)點(diǎn)，降低節(jié)點(diǎn)間負(fù)載差異；

(3)提出一種任務(wù)調(diào)度策略，在降低關(guān)鍵路徑上節(jié)點(diǎn)過高負(fù)載的同時(shí)，最小化關(guān)鍵任務(wù)的節(jié)點(diǎn)間通信開銷，即降低關(guān)鍵邊的通信開銷，實(shí)現(xiàn)計(jì)算資源的最大化利用。

2 關(guān)鍵路徑檢測(cè)算法

本節(jié)從Flink拓?fù)溥壿嬆Ｐ秃臀锢砟Ｐ涂紤]，確定任務(wù)執(zhí)行拓?fù)潢P(guān)系圖中關(guān)鍵路徑，在關(guān)鍵路徑的基礎(chǔ)上，建立關(guān)鍵節(jié)點(diǎn)負(fù)載均衡模型和關(guān)鍵邊最優(yōu)通信開銷模型，在關(guān)鍵路徑上降低部分關(guān)鍵節(jié)點(diǎn)過高負(fù)載的同時(shí)減少關(guān)鍵邊的通信開銷，從而降低整個(gè)任務(wù)拓?fù)鋱?zhí)行的響應(yīng)時(shí)間，為任務(wù)調(diào)度策略的設(shè)計(jì)與實(shí)現(xiàn)提供理論依據(jù)。

定義1 有向無環(huán)圖(DAG)。定義有向無環(huán)圖G=(V(G),E(G)), 其中V(G)={v1,v2,…,vn} 是拓?fù)渲械墓?jié)點(diǎn)集合，E(G)={e,e,…,e} 是拓?fù)渲械挠邢蜻吋?。如果存在e∈E(G)，vk,vj∈V(G)，且vk≠vj，那么表示數(shù)據(jù)流從頂點(diǎn)vk流向頂點(diǎn)vj。有向無環(huán)圖描述了數(shù)據(jù)流的計(jì)算過程，其中圓形為計(jì)算節(jié)點(diǎn)，箭頭表示數(shù)據(jù)的流動(dòng)方向。有向無環(huán)圖如圖1所示。

圖1 拓?fù)溆邢驘o環(huán)圖

定義2 關(guān)聯(lián)度(association degree)。對(duì)于任意節(jié)點(diǎn)vk，若存在以節(jié)點(diǎn)vk為弧尾的n條有向邊e和以vk為弧頭的m條有向邊e，則稱弧vk→vp和弧vq→vk為節(jié)點(diǎn)vk的關(guān)聯(lián)度AD(e,e)。

節(jié)點(diǎn)vk的入度(in-degree)為指向vk的有向邊條數(shù)，記為ID(vi)。出度(out-degree)為從vk指向其它節(jié)點(diǎn)的有向邊條數(shù)，記為OD(vi)。例如圖1中的拓?fù)溆邢驘o環(huán)圖中，存在頂點(diǎn)集合V={va,vb,vc,vd,ve,vf,vg}，有向邊集E={e,e,…,e,e}，源點(diǎn)為va，匯點(diǎn)為vg。則節(jié)點(diǎn)ve的入度為2，出度為1。

定義3 帶權(quán)路徑(path of weight，PW)。定義路徑p(vi,vj)，且∈V(G) 是有向無環(huán)圖中一條從節(jié)點(diǎn)vi開始到節(jié)點(diǎn)vj結(jié)束的一條有向路徑，定義路徑上有向邊的權(quán)重為W={w1,w2,…,wn}。 如果?k，e∈p(vi,vj)，e∈p(vi,vj)，對(duì)于路徑p(vi,vj) 中的任何有向邊e，當(dāng)k≠u時(shí)，?x，e∈p(vi,vj)；當(dāng)u≠v時(shí)，?x，e∈p(vi,vj)。

根據(jù)流式計(jì)算的拓?fù)浣Y(jié)構(gòu)可知，在DAG拓?fù)渲袕脑袋c(diǎn)va到匯點(diǎn)vg存在m條路徑，既P={p1(va,vg),p2(va,vg),…,pm(va,vg)}。 如果將節(jié)點(diǎn)vi流向節(jié)點(diǎn)vk的數(shù)據(jù)流大小作為弧vi→vk的權(quán)重，那么可以將流式計(jì)算拓?fù)鋱D轉(zhuǎn)為帶權(quán)值的AoE-網(wǎng)。

AoE-網(wǎng)拓?fù)淠Ｐ腿鐖D2所示。

圖2 AoE-網(wǎng)拓?fù)淠Ｐ?/p>

由上可知，每條路徑上的計(jì)算延遲是由節(jié)點(diǎn)的計(jì)算開銷和節(jié)點(diǎn)間的通信開銷共同決定的，因此

(1)

其中，cvi表示某條路徑上節(jié)點(diǎn)的計(jì)算延遲，cej表示節(jié)點(diǎn)間的通信延遲。

那么關(guān)鍵路徑可以表示為

Dpj=max{dp1(vs,vt),dp2(vs,vt),…,dpm(vs,vt)}

(2)

根據(jù)定義1、定義2、定義3可知，拓?fù)渲许旤c(diǎn)的最早發(fā)生時(shí)間ve(i)，頂點(diǎn)的最晚發(fā)生時(shí)間vl(j)；在拓?fù)溆邢蜻卐上，存在最早開始時(shí)間e(k)和最晚開始時(shí)間l(k)，當(dāng)e(i)=l(i) 時(shí)，表示這條有向邊e在關(guān)鍵路徑上，并且該有向邊e的權(quán)值為w(e)，表示任務(wù)持續(xù)時(shí)間。

因此，按照拓?fù)漤樞?，拓?fù)渲许旤c(diǎn)的最早發(fā)生時(shí)間ve(i)為

(3)

其中，s為源點(diǎn)，源點(diǎn)的最早發(fā)生時(shí)間為零；E(k)是從節(jié)點(diǎn)i到達(dá)節(jié)點(diǎn)j的所有有向邊的集合。

當(dāng)按照逆拓?fù)漤樞驎r(shí)，拓?fù)渲许旤c(diǎn)的最晚發(fā)生時(shí)間vl(j)為

(4)

其中，t為匯點(diǎn)，匯點(diǎn)的最晚發(fā)生時(shí)間和最早發(fā)生時(shí)間相等；E(k)是從節(jié)點(diǎn)i發(fā)出的所有有向邊的集合。

拓?fù)渲忻織l有向邊e的最早開始時(shí)間e(k)為

e(k)=ve(j)

(5)

拓?fù)渲杏邢蜻卐的最晚開始時(shí)間l(k)為

l(k)=vl(j)-w(e)

(6)

算法1：關(guān)鍵路徑檢測(cè)算法(CP-Algorithm)。

輸入：有向無環(huán)圖G=，

有向邊權(quán)值集合W←{w1,w2,…,wn}；

輸出：關(guān)鍵節(jié)點(diǎn)集合Vcp，關(guān)鍵邊集合Ecp；

(1) if ID(vi)=0；/*從源點(diǎn)s出發(fā)進(jìn)行遍歷*/

(2) for i=1;i≤n-1;i++；

(3) ve[i+1]←ve[i]+wi；/*計(jì)算vertex的最早發(fā)生時(shí)間*/

(4) vl[n-1]←ve[n-1]；

(5) if OD(vi)=0；/*從匯點(diǎn)t出發(fā)進(jìn)行遍歷*/

(6) for j=n-2;j>1;j--；

(7) vl[i-1]←vl[i]-wi；/*計(jì)算vertex的最晚發(fā)生時(shí)間*/

(8)通過式(3)和式(4)計(jì)算最早開始時(shí)間e和最晚開始時(shí)間l；

(9)when e=l；/*判斷當(dāng)前有向邊為關(guān)鍵邊*/

(11) end；

在算法1中，對(duì)于Flink環(huán)境中的DAG將數(shù)據(jù)流大小作為有向邊的權(quán)重構(gòu)建AoE-網(wǎng)，然后CP-Algorithm依次對(duì)數(shù)據(jù)拓?fù)銩oE網(wǎng)進(jìn)行正向和反向遍歷，通過步驟(2)～步驟 (9)確定數(shù)據(jù)拓?fù)渲嘘P(guān)鍵節(jié)點(diǎn)和關(guān)鍵邊，因此，該算法的時(shí)間復(fù)雜度為T(n)=O(n+e)，且在空間復(fù)雜度上，DAG拓?fù)浣Y(jié)構(gòu)并未發(fā)生改變，因此，該算法是可行的。

3 關(guān)鍵路徑模型定義及算法

在本章節(jié)中通過算法1中檢測(cè)到的關(guān)鍵節(jié)點(diǎn)集合和關(guān)鍵邊集合對(duì)問題進(jìn)行定義和建模。

3.1 關(guān)鍵節(jié)點(diǎn)負(fù)載均衡模型

(7)

(8)

(9)

由上可知，式(7)表示理想狀態(tài)下關(guān)鍵節(jié)點(diǎn)的負(fù)載情況，式(9)表示關(guān)鍵節(jié)點(diǎn)的實(shí)際權(quán)重與理想權(quán)重的偏離程度，并且標(biāo)準(zhǔn)差越小表示各個(gè)工作節(jié)點(diǎn)的負(fù)載偏離度越低，負(fù)載越趨于均衡。

3.2 關(guān)鍵邊最優(yōu)通信開銷模型

圖3 任務(wù)分配模型

如上所述，在關(guān)鍵路徑上存在節(jié)點(diǎn)間通信和節(jié)點(diǎn)內(nèi)通信，且節(jié)點(diǎn)間通信開銷遠(yuǎn)大于節(jié)點(diǎn)內(nèi)通信開銷，通過將節(jié)點(diǎn)間的通信開銷盡可能地轉(zhuǎn)化為節(jié)點(diǎn)內(nèi)通信開銷，能夠降低關(guān)鍵路徑響應(yīng)時(shí)間，從而降低整個(gè)任務(wù)拓?fù)涞捻憫?yīng)時(shí)間?；谝陨纤枷?，提出定理1。

定理1 最優(yōu)通信開銷定理。當(dāng)關(guān)鍵路徑上不存在或節(jié)點(diǎn)間通信開銷最少時(shí)，最小化的節(jié)點(diǎn)間通信開銷等價(jià)于最大化的節(jié)點(diǎn)內(nèi)通信開銷。即

(10)

證明：由Flink拓?fù)淠Ｐ涂芍?，?dāng)提交拓?fù)浣o節(jié)點(diǎn)后，拓?fù)鋵?shí)例便不會(huì)發(fā)生改變，其包含的任務(wù)總數(shù)和數(shù)據(jù)流總數(shù)不可改變。因此，設(shè)總數(shù)據(jù)流大小為定值R，即

(11)

證畢。

對(duì)于節(jié)點(diǎn)內(nèi)線程間通信開銷，F(xiàn)link提供SlotSharingGroup類，會(huì)盡可能地讓更多的子任務(wù)共享一個(gè)任務(wù)槽；提供ColocationGroup類可將子任務(wù)強(qiáng)制放入一個(gè)任務(wù)槽內(nèi)，SlotSharingGroup類和ColocationGroup類這兩種方法為我們減少關(guān)鍵節(jié)點(diǎn)內(nèi)線程間通信開銷提供了幫助。對(duì)于節(jié)點(diǎn)間通信，通過Flink提供的operator chains，會(huì)盡可能地將operator的子任務(wù)chain在一起形成一個(gè)任務(wù)，每個(gè)任務(wù)在一個(gè)線程中執(zhí)行，通過設(shè)置operator chains能夠減少進(jìn)程之間的切換，減少進(jìn)程之間通信開銷。因此，對(duì)于節(jié)點(diǎn)間通信，為達(dá)到定理1關(guān)鍵邊最優(yōu)通信開銷模型要求，盡可能地將節(jié)點(diǎn)間通信轉(zhuǎn)為節(jié)點(diǎn)內(nèi)通信方式，并且在降低關(guān)鍵節(jié)點(diǎn)計(jì)算開銷的同時(shí)降低關(guān)鍵邊的通信開銷，即在保證關(guān)鍵節(jié)點(diǎn)負(fù)載差異較小的同時(shí)降低關(guān)鍵節(jié)點(diǎn)間的通信開銷，盡可能地將負(fù)載過高關(guān)鍵節(jié)點(diǎn)上的任務(wù)調(diào)度到負(fù)載較低的計(jì)算節(jié)點(diǎn)上。

3.3 基于拓?fù)潢P(guān)鍵路徑的任務(wù)調(diào)度算法

基于關(guān)鍵路徑的任務(wù)調(diào)度算法主要是在保證系統(tǒng)性能不發(fā)生改變的情況下，盡可能使得各關(guān)鍵節(jié)點(diǎn)負(fù)載差異較小的同時(shí)減少關(guān)鍵邊的通信開銷，從而降低整個(gè)任務(wù)拓?fù)鋱?zhí)行響應(yīng)時(shí)間，實(shí)現(xiàn)資源最大化利用。在上一節(jié)中通過關(guān)鍵路徑檢測(cè)算法確定拓?fù)潢P(guān)鍵路徑，并獲取關(guān)鍵路徑上權(quán)重集合Wcp、節(jié)點(diǎn)集合Vcp、邊集合Ecp。并且通過負(fù)載模型判斷出負(fù)載較高的關(guān)鍵節(jié)點(diǎn)，對(duì)此負(fù)載較高節(jié)點(diǎn)上存在節(jié)點(diǎn)間通信的任務(wù)執(zhí)行任務(wù)遷移策略，將該任務(wù)的節(jié)點(diǎn)間通信開銷轉(zhuǎn)為節(jié)點(diǎn)內(nèi)通信開銷，在保證關(guān)鍵節(jié)點(diǎn)負(fù)載差異較小的同時(shí)降低任務(wù)的通信開銷。

為了達(dá)到上述關(guān)鍵節(jié)點(diǎn)負(fù)載均衡模型和關(guān)鍵邊最優(yōu)通信開銷模型的要求，提出了一種在Flink環(huán)境下的任務(wù)調(diào)度策略(TSS-Flink)。其算法具體過程如下：

算法2：拓?fù)潢P(guān)鍵路徑上任務(wù)調(diào)度算法。

(1)quicksort(Wcp，DESC)；

/* 對(duì)輸入的關(guān)鍵邊權(quán)重集合元素降序排序 */

(3)calculate theδby (9)；

/* 判斷關(guān)鍵路徑上是否存在負(fù)載不均衡的節(jié)點(diǎn) */

/* 確定不均衡節(jié)點(diǎn)以及該節(jié)點(diǎn)上任務(wù)和前驅(qū)任務(wù)的集合 */

/* 確定關(guān)鍵節(jié)點(diǎn)上任務(wù)和它的前驅(qū)任務(wù) */

(8) if np≠nq

(12) reschedule CP-Algorithm；

(14)end while；

4 實(shí)驗(yàn)與分析

Apache Flink 作為開源免費(fèi)的分布式數(shù)據(jù)流處理平臺(tái)之一，在實(shí)時(shí)業(yè)務(wù)中得到廣泛應(yīng)用。對(duì)于本章節(jié)的實(shí)驗(yàn)，通過在Flink平臺(tái)上實(shí)現(xiàn)TSS-Flink策略，對(duì)該策略的有效性進(jìn)行驗(yàn)證。

4.1 實(shí)驗(yàn)環(huán)境

實(shí)驗(yàn)環(huán)境是由7個(gè)相同配置的普通物理PC機(jī)組成的Flink集群，其中包含1個(gè)JobManager節(jié)點(diǎn)，該節(jié)點(diǎn)負(fù)責(zé)Flink集群的作業(yè)調(diào)度和資源管理；6個(gè)TaskManager節(jié)點(diǎn)，負(fù)責(zé)執(zhí)行具體任務(wù)計(jì)劃。此外，配置1個(gè)Zookeeper節(jié)點(diǎn)負(fù)責(zé)在任務(wù)執(zhí)行過程中監(jiān)控和記錄數(shù)據(jù)節(jié)點(diǎn)；1個(gè)Kafka節(jié)點(diǎn)和1個(gè)HDFS節(jié)點(diǎn)作為數(shù)據(jù)流的源點(diǎn)和匯點(diǎn)。各節(jié)點(diǎn)的具體的分布情況見表1。

表1 Flink集群節(jié)點(diǎn)分布

在Flink集群環(huán)境中，為保證實(shí)驗(yàn)的順利進(jìn)行，集群均采用相同的配置，具體配置參數(shù)見表2。

為了使TSS-Flink算法達(dá)到最優(yōu)的執(zhí)行效果，通過多次對(duì)原系統(tǒng)反復(fù)實(shí)驗(yàn)，最終確定實(shí)驗(yàn)相關(guān)參數(shù)。配置情況見表3。

在本章節(jié)的實(shí)驗(yàn)測(cè)試中，執(zhí)行了Streaming Benchmark 中的WordCount、TwitterSentiment基準(zhǔn)測(cè)試進(jìn)行驗(yàn)證。在WordCount基準(zhǔn)測(cè)試中，以英文小說《Harry Potter》作為

表2 節(jié)點(diǎn)配置信息

表3 TSS-Flink算法參數(shù)設(shè)置

輸入數(shù)據(jù)源，統(tǒng)計(jì)單詞頻次，其計(jì)算復(fù)雜度相對(duì)較低但對(duì)CPU資源的占用率較高。TwitterSentiment是一個(gè)針對(duì)Twitter用戶所發(fā)的推文內(nèi)容進(jìn)行情感分析的作業(yè)，該作業(yè)以160 000條文本作為輸入數(shù)據(jù)源，其對(duì)內(nèi)存資源和CPU資源占用相對(duì)都較高。通過以上兩個(gè)基準(zhǔn)測(cè)試，能夠?qū)SS-Flink算法的有效性進(jìn)行驗(yàn)證。

4.2 實(shí)驗(yàn)分析

本章節(jié)中通過執(zhí)行WordCount和TwitterSentiment這兩組資源敏感型基準(zhǔn)測(cè)試，從計(jì)算延遲、CPU負(fù)載和RAM占用率3個(gè)方面對(duì)Flink集群中各個(gè)工作節(jié)點(diǎn)進(jìn)行性能監(jiān)測(cè)和評(píng)估，以驗(yàn)證TSS-Flink的優(yōu)化效果。

本節(jié)討論基準(zhǔn)測(cè)試WordCount作業(yè)在Apache Flink默認(rèn)調(diào)度算法和TSS-Flink下分別運(yùn)行時(shí)集群各工作節(jié)點(diǎn)的負(fù)載情況。由于Flink默認(rèn)調(diào)度算法采用隨機(jī)的方式分配任務(wù)，當(dāng)從Source operator發(fā)送數(shù)據(jù)流到Sink operator時(shí)，極易導(dǎo)致各工作節(jié)點(diǎn)資源分配不均、負(fù)載差異較大情況，且TSS-Flink算法在執(zhí)行過程中應(yīng)該考慮到任務(wù)分配所導(dǎo)致的負(fù)載差異性。從圖4所示的實(shí)驗(yàn)結(jié)果中可以得出：在Flink默認(rèn)調(diào)度算法下，各個(gè)節(jié)點(diǎn)的CPU負(fù)載不均衡且差異較大，其中負(fù)載最高的節(jié)點(diǎn)是node5，負(fù)載最低的節(jié)點(diǎn)是node6，節(jié)點(diǎn)之間CPU負(fù)載最大相差28%。當(dāng)節(jié)點(diǎn)node2和節(jié)點(diǎn)node5的CPU負(fù)載超過表3中設(shè)置的閾值0.7時(shí)觸發(fā)TSS-Flink算法，該算法執(zhí)行后集群中各工作節(jié)點(diǎn)的CPU負(fù)載差異明顯縮小且均低于用戶設(shè)置閾值0.7，且其執(zhí)行后的CPU負(fù)載標(biāo)準(zhǔn)差比Flink默認(rèn)調(diào)度算法降低了8.28%。通過對(duì)集群各工作節(jié)點(diǎn)的負(fù)載均衡測(cè)試驗(yàn)證了TSS-Flink算法的有效性。

圖4 WordCount CPU負(fù)載對(duì)比

為了進(jìn)一步驗(yàn)證TSS-Flink策略的優(yōu)化效果，在本章節(jié)中繼續(xù)對(duì)benchmark作業(yè)執(zhí)行過程實(shí)時(shí)監(jiān)控節(jié)點(diǎn)的內(nèi)存占用率。在Flink中，通過Monitor模塊進(jìn)行內(nèi)存實(shí)時(shí)監(jiān)控，定義OperatorScopeFormat.java類獲取System Metrics。在實(shí)時(shí)監(jiān)控過程中，通過定點(diǎn)采樣得到如圖5所示的實(shí)驗(yàn)結(jié)果：當(dāng)單位時(shí)間內(nèi)數(shù)據(jù)元組數(shù)量不斷增加時(shí)，原系統(tǒng)部分節(jié)點(diǎn)由于負(fù)載過高導(dǎo)致內(nèi)存占用率急劇上升，并且這些負(fù)載較高的節(jié)點(diǎn)無法及時(shí)處理數(shù)據(jù)從而導(dǎo)致拓?fù)涮幚頃r(shí)延變長(zhǎng)，而另外一部分節(jié)點(diǎn)的資源也無法得到充分利用。通過使用TSS-Flink策略，對(duì)負(fù)載較高的節(jié)點(diǎn)上的任務(wù)重調(diào)度，使得拓?fù)浞顷P(guān)鍵路徑上的節(jié)點(diǎn)分擔(dān)拓?fù)潢P(guān)鍵路徑上負(fù)載過高節(jié)點(diǎn)的資源使用壓力，最終被采樣節(jié)點(diǎn)的內(nèi)存利用率都有一定程度的下降且逐步趨于平穩(wěn)狀態(tài)。

圖5 WordCount內(nèi)存占用對(duì)比

圖6表示benchmark在Flink默認(rèn)調(diào)度算法和TSS-Flink下的工作節(jié)點(diǎn)間通信開銷，不管是在默認(rèn)調(diào)度算法還是TSS-Flink下，節(jié)點(diǎn)間數(shù)據(jù)流大小均經(jīng)歷一個(gè)從0快速上升到正常狀態(tài)的過程。TSS-Flink算法在執(zhí)行中將關(guān)鍵節(jié)點(diǎn)上的線程遷移至前驅(qū)非關(guān)鍵節(jié)點(diǎn)上，從而減少線程節(jié)點(diǎn)間通信開銷。Flink默認(rèn)調(diào)度算法運(yùn)行且趨于穩(wěn)定后(90 s-300 s)，節(jié)點(diǎn)間數(shù)據(jù)流大小的平均值約為16 572 tuples/s；當(dāng)執(zhí)行TSS-Flink算法且系統(tǒng)趨于穩(wěn)定后(125 s-250 s)，節(jié)點(diǎn)間數(shù)據(jù)流大小約為12 410 tuples/s，相比Flink默認(rèn)調(diào)度算法降低了25.1%?？梢?，TSS-Flink在降低節(jié)點(diǎn)間通信開銷方面具有更為明顯的效果且符合最優(yōu)通信開銷模型思想，也進(jìn)一步驗(yàn)證了算法的有效性。

圖6 節(jié)點(diǎn)間數(shù)據(jù)流大小對(duì)比

圖7表示任務(wù)拓?fù)渲袇R點(diǎn)接收從source發(fā)出的每 10 000 條tuples時(shí)記錄一個(gè)延遲時(shí)間并持續(xù)15 min得到的實(shí)驗(yàn)結(jié)果，實(shí)驗(yàn)在WordCount作業(yè)上執(zhí)行TSS算法與原系統(tǒng)算法比較得出：因?yàn)閃ordCount作業(yè)復(fù)雜度低于Twitter作業(yè)復(fù)雜度，所以WordCount作業(yè)的計(jì)算延遲相對(duì)較低，當(dāng)經(jīng)過TSS-Flink算法進(jìn)行優(yōu)化后，系統(tǒng)的計(jì)算延遲明顯下降。在原系統(tǒng)中，隨著數(shù)據(jù)流的連續(xù)不斷輸入，計(jì)算延遲也隨著慢慢升高，當(dāng)某些節(jié)點(diǎn)的計(jì)算資源達(dá)到瓶頸無法及時(shí)處理數(shù)據(jù)時(shí)，導(dǎo)致計(jì)算延遲過長(zhǎng)，系統(tǒng)執(zhí)行任務(wù)拓?fù)涞膶?shí)時(shí)性較差。通過TSS-Flink算法，對(duì)計(jì)算資源相對(duì)緊張的關(guān)鍵節(jié)點(diǎn)上的任務(wù)執(zhí)行調(diào)度策略，對(duì)比原系統(tǒng)該策略使集群的計(jì)算延遲降低最多達(dá)到388 ms，至少40 ms，平均降低了248 ms。

圖7 WordCount計(jì)算延遲對(duì)比

對(duì)于圖8，執(zhí)行的Twitter作業(yè)本身復(fù)雜度比WordCount作業(yè)較高，因此計(jì)算延遲也較高。通過在連續(xù)15 min內(nèi)記錄匯點(diǎn)每接受10 000條tuples數(shù)據(jù)時(shí)的延遲時(shí)間，可以得出：原系統(tǒng)在數(shù)據(jù)流不斷增加和快速變化下，導(dǎo)致數(shù)據(jù)流無法及時(shí)處理，造成數(shù)據(jù)堆積，數(shù)據(jù)堆積節(jié)點(diǎn)延遲過高，進(jìn)而影響系統(tǒng)的實(shí)時(shí)計(jì)算能力。通過執(zhí)行TSS-Flink算法將任務(wù)拓?fù)渲忻?0 000條tuples數(shù)據(jù)的計(jì)算延遲最多降低了210 ms，至少降低了8 ms，平均降低了130 ms，該調(diào)度策略有效地降低了節(jié)點(diǎn)間通信開銷和關(guān)鍵路徑上計(jì)算延遲，提高了計(jì)算的實(shí)時(shí)性，使計(jì)算資源達(dá)到最大化利用。

圖8 Twitter計(jì)算延遲對(duì)比

綜上所述，實(shí)驗(yàn)驗(yàn)證TSS-Flink算法對(duì)夠通過降低節(jié)點(diǎn)間通信開銷從而降低響應(yīng)時(shí)間，提高集群的性能。通過圖7、圖8可知，不同的作業(yè)類型下該策略對(duì)系統(tǒng)的計(jì)算延遲優(yōu)化效果并不相同，但其平均優(yōu)化比提高了13.09%，有效地降低了計(jì)算延遲，提高了系統(tǒng)性能。

5 結(jié)束語

通過對(duì)比現(xiàn)有的任務(wù)調(diào)度算法，發(fā)現(xiàn)多是對(duì)負(fù)載較重的節(jié)點(diǎn)執(zhí)行任務(wù)調(diào)度策略，雖然這些調(diào)度策略能有效降低任務(wù)拓?fù)漤憫?yīng)時(shí)間，提高系統(tǒng)性能。但也并未考慮到各任務(wù)的計(jì)算開銷和任務(wù)之間的通信開銷，且并未在Apache Flink平臺(tái)上實(shí)現(xiàn)該任務(wù)調(diào)度策略，所以在節(jié)點(diǎn)間負(fù)載均衡和通信開銷方面仍然存在很大的優(yōu)化空間。本文通過找到直接影響整個(gè)任務(wù)拓?fù)漤憫?yīng)時(shí)間的關(guān)鍵路徑，確定負(fù)載較高的關(guān)鍵節(jié)點(diǎn)和該節(jié)點(diǎn)上通信開銷較大的關(guān)鍵邊，建立關(guān)鍵節(jié)點(diǎn)負(fù)載均衡模型和關(guān)鍵邊最優(yōu)通信開銷模型，提出一種Flink環(huán)境下的任務(wù)調(diào)度策略(TSS-Flink)。通過WordCount和Twitter兩個(gè)benchmark的實(shí)驗(yàn)驗(yàn)證，結(jié)果表明算法能夠?qū)崿F(xiàn)對(duì)Flink集群的性能優(yōu)化，盡可能地更好地利用計(jì)算資源。

下一步的研究工作將重點(diǎn)關(guān)注由于輸入數(shù)據(jù)流的急劇變化造成的資源分配不均問題，針對(duì)關(guān)鍵路徑上的負(fù)載傾斜較為嚴(yán)重的關(guān)鍵節(jié)點(diǎn)，如何判斷出節(jié)點(diǎn)內(nèi)的任務(wù)通過橫向遷移和縱向遷移實(shí)現(xiàn)資源的最大化利用且保證遷移后的拓?fù)浣Y(jié)構(gòu)不發(fā)生改變。