李文佳，史 嵐，季航旭，羅意彭

(1.東北大學(xué)計算機科學(xué)與工程學(xué)院，遼寧 沈陽 110169；2.遼寧工業(yè)大學(xué)軟件學(xué)院，遼寧 錦州 121000)

1 引言

近年來，隨著數(shù)據(jù)經(jīng)濟在全球的加速推進，以及5G、人工智能、物聯(lián)網(wǎng)等相關(guān)技術(shù)的快速發(fā)展，全球數(shù)據(jù)量迎來巨大規(guī)模的爆發(fā)，越來越多的政府機構(gòu)和研究人員開始重視這種大量數(shù)據(jù)的收集、使用與處理。伴隨著數(shù)據(jù)的爆炸式增長與多種多樣需求的出現(xiàn)，一些傳統(tǒng)的大數(shù)據(jù)模型和分布式計算引擎已經(jīng)很難滿足當(dāng)前業(yè)務(wù)的需求，因此，許多新的分布式計算框架應(yīng)運而生。大數(shù)據(jù)計算引擎的發(fā)展歷程主要分為4個階段。第一代大數(shù)據(jù)計算引擎是谷歌于2004年提出的基于MapReduce[1]的Hadoop[2]計算引擎。Hadoop主要依靠把任務(wù)拆分成map和reduce 2個階段去處理，這種模式由于難以支持迭代計算，因此產(chǎn)生了第二代基于有向無環(huán)圖DAG(Directed Acyclic Graph)[3]的以Tez[4]和Oozie為代表的計算引擎。雖然第二代計算引擎解決了MapReduce中不支持迭代計算的問題，但是由于這種計算引擎只能處理離線任務(wù)，在線任務(wù)處理需求增加的驅(qū)動下，產(chǎn)生了第三代基于彈性分布式數(shù)據(jù)集RDD(Resilient Distributed Dataset)[5]的Spark[6]計算引擎。Spark既可以處理離線計算也可以處理實時計算，它是在Tez的基礎(chǔ)上對Job作了更細粒度的拆分，但是其延遲較大，難以處理實時需求更高的連續(xù)流數(shù)據(jù)請求。因此，產(chǎn)生了現(xiàn)在主流的可以處理高實時性任務(wù)的第四代大數(shù)據(jù)計算引擎Flink[7]。Flink對事件時間的支持、精確一次(Exactly-Once)的狀態(tài)一致性以及內(nèi)部檢查點機制等特性，決定了其在大數(shù)據(jù)計算引擎上占據(jù)主流地位?，F(xiàn)如今越來越多的公司采用基于Flink的大數(shù)據(jù)計算引擎去實現(xiàn)多種場景，比如阿里巴巴雙十一實時大屏的投放、騰訊實時平臺的搭建以及美團、餓了么、愛奇藝等公司的數(shù)據(jù)處理流程都是基于Flink構(gòu)建的。

在大數(shù)據(jù)計算引擎Flink中，大量的計算任務(wù)需要被調(diào)度到資源節(jié)點上，如何使整體任務(wù)用最少的完成時間，在很大程度上由它的調(diào)度算法決定，因此良好的任務(wù)調(diào)度算法是分布式計算的重要組成部分。有效的任務(wù)調(diào)度是分布式計算的一個關(guān)鍵問題，其目標(biāo)是在滿足任務(wù)依賴關(guān)系的前提下，調(diào)整任務(wù)的執(zhí)行順序，將任務(wù)分配給對應(yīng)的資源，使整個系統(tǒng)的任務(wù)能在最短的時間內(nèi)執(zhí)行完成。

由于集群的異構(gòu)性以及不同算子復(fù)雜度不同，大數(shù)據(jù)計算系統(tǒng)中不可避免地會出現(xiàn)負(fù)載不均的情況，本文提出了基于資源反饋的負(fù)載均衡任務(wù)調(diào)度算法RFTS(load balancing Task Scheduling algorithm based on Resource Feedback)。與傳統(tǒng)的負(fù)載均衡算法不同的是，RFTS算法綜合考慮了集群計算資源的實時負(fù)載情況以及處理任務(wù)的優(yōu)先級和順序，更高效地完成任務(wù)與計算資源之間的分配，通過實時資源監(jiān)控、區(qū)域劃分和基于人工螢火蟲優(yōu)化GSO(Glowworm Swarm Optimization)的任務(wù)調(diào)度算法3個模塊，把負(fù)載過重的機器中處于等待隊列中的任務(wù)分配給負(fù)載較輕的機器，提高系統(tǒng)處理任務(wù)的執(zhí)行效率和集群利用率。

本文的主要貢獻包括以下3個方面:

(1) 設(shè)計了一個Flink系統(tǒng)的實時監(jiān)控系統(tǒng)Monitor，實時監(jiān)控每個從節(jié)點(Slave)的CPU核數(shù)、CPU利用率、內(nèi)存利用率和總內(nèi)存大小等性能指標(biāo)，從而獲取每個資源節(jié)點的負(fù)載大小。

(2) 提出基于資源反饋的負(fù)載均衡任務(wù)調(diào)度算法。該算法在集群出現(xiàn)負(fù)載不均時，重新分配每個資源節(jié)點的任務(wù)，以提高系統(tǒng)執(zhí)行效率和整體資源利用率。該算法通過實現(xiàn)的實時監(jiān)控系統(tǒng)Monitor來監(jiān)控資源節(jié)點的負(fù)載情況，并根據(jù)區(qū)域劃分算法把集群劃分為過負(fù)載、輕負(fù)載、近飽和和差飽和4個區(qū)域，由于過負(fù)載區(qū)域的機器負(fù)載過重會影響整個集群的執(zhí)行效率，因此用基于人工螢火蟲優(yōu)化算法的調(diào)度策略，把過負(fù)載區(qū)域中資源節(jié)點位于等待隊列的任務(wù)調(diào)度給差飽和區(qū)域的資源節(jié)點。

(3) 通過編寫源碼，在大數(shù)據(jù)計算系統(tǒng)Flink中，實現(xiàn)了RFTS算法。

最后在TPC-C和TPC-H[8]數(shù)據(jù)集上對基于資源反饋的負(fù)載均衡任務(wù)調(diào)度算法進行了實驗，實驗結(jié)果表明，該算法在執(zhí)行時間和吞吐量方面均有明顯的提升效果．

2 相關(guān)工作

任務(wù)調(diào)度是指系統(tǒng)將用戶提交的任務(wù)通過某種方式進行拆分、重組并分配到集群中對應(yīng)的資源節(jié)點進行計算的過程。眾所周知，任務(wù)調(diào)度問題是NP-hard[9]。大數(shù)據(jù)計算模型是一種新型的分布式計算模型，專門用于處理海量數(shù)據(jù)的存儲、分析和計算，其優(yōu)點在于能以使用較低的時間和空間成本來實現(xiàn)系統(tǒng)的高可擴性和可伸縮性。這些決定了大數(shù)據(jù)計算模型不僅可以應(yīng)對數(shù)據(jù)日益增長的計算、存儲和分析需求，也可以很好地滿足并適應(yīng)網(wǎng)絡(luò)環(huán)境中復(fù)雜多變的特點，保障基本的網(wǎng)絡(luò)性能請求。大數(shù)據(jù)計算中資源的服務(wù)質(zhì)量好壞是衡量大數(shù)據(jù)計算效果的一個重要方面，但是大數(shù)據(jù)計算中云端存在諸多形態(tài)，且系統(tǒng)規(guī)模巨大，資源節(jié)點之間的結(jié)構(gòu)差異性也較大，因此如何更好地實現(xiàn)任務(wù)調(diào)度就成為了大數(shù)據(jù)計算研究中的熱點和難點問題。

調(diào)度算法的目的是將任務(wù)調(diào)度到資源節(jié)點的同時使得任務(wù)的執(zhí)行時間和吞吐量盡可能好。任務(wù)執(zhí)行所需要的資源、網(wǎng)絡(luò)IO、耗費的時間和用戶需求等都由任務(wù)調(diào)度策略決定。因此,在分布式計算中任務(wù)調(diào)度很大程度上決定了分布式計算系統(tǒng)的系統(tǒng)性能。

分布式計算中常用的調(diào)度算法有經(jīng)典調(diào)度算法和啟發(fā)式調(diào)度算法。經(jīng)典調(diào)度算法主要有Max-Min算法[10]、Max-Max算法和Sufferage算法[11]、先到先服務(wù)、輪詢調(diào)度算法[12]以及公平調(diào)度算法[13]等,這些算法由于參數(shù)少、操作簡單、容易復(fù)現(xiàn)等優(yōu)點被廣泛使用。但是，這類算法也存在著面對復(fù)雜場景和數(shù)據(jù)頻繁被調(diào)度的場景時任務(wù)分配效果較差、容易導(dǎo)致數(shù)據(jù)傾斜等缺點[14]。因此，研究人員提出了主要針對復(fù)雜場景的啟發(fā)式調(diào)度算法。Holland于1975年通過觀察生物界進化的規(guī)律，通過組合交叉、遺傳和變異的方式提出了遺傳算法[15];1991年意大利學(xué)者Dorigo通過模擬蟻群根據(jù)信息素的反饋信息不斷改變尋找食物的速度和方向的行為提出了蟻群算法[16]；1995年Eberhart和Kenny博士通過觀察鳥群在遷移過程中根據(jù)其他鳥類的飛行軌跡來改變自身速度和位置的行為模式，提出了粒子群算法[17]；還有現(xiàn)在被廣泛使用的模擬退火算法[18]、差分演化算法[19]等。啟發(fā)式調(diào)度算法雖然適合復(fù)雜場景并能夠快速展開全局搜索，但也存在隨機性過高、容易陷入局部最優(yōu)解和參數(shù)難以控制等缺點。

Flink系統(tǒng)中默認(rèn)的調(diào)度策略是輪詢調(diào)度算法，這種算法不考慮每個資源節(jié)點的異構(gòu)性，容易導(dǎo)致集群負(fù)載不均，使性能強的資源節(jié)點和性能差的資源節(jié)點面對同樣多的任務(wù)，這時性能差的資源節(jié)點需要更多的時間，根據(jù)水桶效應(yīng)，系統(tǒng)的執(zhí)行效率是由性能最差的節(jié)點所決定的，因此集群負(fù)載不均會降低整個系統(tǒng)的執(zhí)行效率。

基于上述分析可知，現(xiàn)有的調(diào)度算法不足以高效優(yōu)化實時大數(shù)據(jù)計算系統(tǒng)Flink的執(zhí)行速度，因此需要研究新的任務(wù)調(diào)度算法，使其在面對Flink系統(tǒng)復(fù)雜場景時能保證負(fù)載均衡且盡可能地提高系統(tǒng)執(zhí)行效率。所以，本文提出了基于資源反饋的負(fù)載均衡任務(wù)調(diào)度算法RFTS。與傳統(tǒng)的負(fù)載均衡調(diào)度算法相比，RFTS算法根據(jù)集群中每臺機器當(dāng)前的負(fù)載壓力和任務(wù)優(yōu)先級來快速而高效地分配任務(wù)，可以在不降低任務(wù)處理實時性的前提下，提高系統(tǒng)處理任務(wù)的總體執(zhí)行效率。

3 基于資源反饋的負(fù)載均衡任務(wù)調(diào)度算法RFTS

由于集群的異構(gòu)性以及不同算子復(fù)雜度不同，分布式計算系統(tǒng)中不可避免地會出現(xiàn)負(fù)載不均的情況。針對該問題，本文提出了基于資源反饋的負(fù)載均衡任務(wù)調(diào)度算法。

在本節(jié)中，首先介紹RFTS算法的基本思想;接著，借助實例分別介紹RFTS算法包含的3個主要模塊。

3.1 定義與說明

首先對RFTS中用到的變量和公式進行定義與說明。令TM={TM1,TM2,…,TMm}為分布式系統(tǒng)中資源節(jié)點的集合，T={T1,T2,…,Tn}為需要執(zhí)行的全部任務(wù)的集合。

定義1(任務(wù)整體完成時間) 任務(wù)整體完成時間是指從第一個任務(wù)執(zhí)行開始到所有任務(wù)中最后一個任務(wù)完成所經(jīng)歷的時間，記為TotalTime。負(fù)載均衡的目標(biāo)是使得TotalTime盡可能小。

定義2(資源節(jié)點性能指標(biāo)) 資源節(jié)點的性能指標(biāo)如式(1)所示:

Metricj=C1*Core_Numj*ω1j+

C2*(1-ω2j)*ω3j,j=1,2,…,m

(1)

其中，C1、C2為常數(shù)，Core_Numj表示資源節(jié)點TMj的CPU核數(shù)，ω1j表示資源節(jié)點TMj的CPU利用率，ω2j表示資源節(jié)點TMj的內(nèi)存利用率，ω3j表示資源節(jié)點TMj的總內(nèi)存大小。因此，資源節(jié)點性能指標(biāo)Metricj由CPU核數(shù)、CPU利用率和空閑內(nèi)存大小決定，最后作歸一化處理。

定義3(資源節(jié)點負(fù)載值) 負(fù)載值代表該資源節(jié)點上任務(wù)負(fù)載量相對于該節(jié)點當(dāng)前性能的承擔(dān)能力。負(fù)載值越小，表示該節(jié)點承受負(fù)載的能力越好，即可以承受更多的任務(wù)；反之，負(fù)載值越大，該節(jié)點承受負(fù)載的能力越差，需要減少該節(jié)點執(zhí)行的任務(wù)。負(fù)載值的定義如式(2)所示:

(2)

其中，TaskNumj表示資源節(jié)點TMj的任務(wù)隊列中的任務(wù)數(shù)量；Metricj代表資源節(jié)點TMj的性能指標(biāo)，Metricj越大，代表該資源節(jié)點性能越好，可以在單位時間內(nèi)處理更多的任務(wù)；反之，則代表該資源節(jié)點的性能越差，在單位時間內(nèi)能夠處理的任務(wù)越少。

集群負(fù)載平均值的定義如式(3)所示：

(3)

定義4(集群負(fù)載值) 集群負(fù)載值表示整個集群的負(fù)載均衡程度，如式(4)所示。如果集群負(fù)載值C_Load小于閾值β，則說明集群整體負(fù)載處于均衡狀態(tài)，算法結(jié)束；否則，集群負(fù)載值越大，說明集群負(fù)載不均的程度越高，此時需要將每個節(jié)點的負(fù)載值和集群負(fù)載值存入MongoDB數(shù)據(jù)庫中，為下一階段的區(qū)域劃分和任務(wù)重新調(diào)度做準(zhǔn)備。

(4)

3.2 主要思想

由于Flink系統(tǒng)中任務(wù)以隨機順序調(diào)度，本文提出了一種基于資源反饋的負(fù)載均衡任務(wù)調(diào)度算法，算法構(gòu)建了一個優(yōu)化器，通過集群資源實時監(jiān)控、集群區(qū)域劃分和任務(wù)重新分配去優(yōu)化Flink系統(tǒng)中的調(diào)度策略，以避免負(fù)載不均的情況。

RFTS算法的總體框架如圖1所示，其基本思想是先通過實時監(jiān)控集群的性能情況，得到每個資源節(jié)點的負(fù)載值和整個集群的負(fù)載值，并把這些信息實時存儲到MongoDB數(shù)據(jù)庫中，集群的負(fù)載越小，則說明整個集群的負(fù)載越均衡。因此，當(dāng)負(fù)載均衡值小于閾值時，說明集群處于均衡狀態(tài)，算法結(jié)束；否則，利用區(qū)域劃分算法劃分整個集群，把集群分成過負(fù)載、輕負(fù)載、近飽和和差飽和4個區(qū)域，最后采用基于人工螢火蟲優(yōu)化的任務(wù)調(diào)度算法結(jié)合任務(wù)的優(yōu)先級把過負(fù)載區(qū)域的任務(wù)遷移到差飽和區(qū)域中的資源節(jié)點上，以降低整個集群的負(fù)載值。

Figure 1 General framework of RFTS圖1 RFTS算法總體框架

本節(jié)進一步優(yōu)化基于Flink本身的調(diào)度算法，針對已經(jīng)完成初步調(diào)度策略的集群進行實時性能監(jiān)控，當(dāng)負(fù)載不均時根據(jù)各節(jié)點任務(wù)隊列中任務(wù)的數(shù)量和任務(wù)的優(yōu)先級重新產(chǎn)生調(diào)度策略。

3.3 資源監(jiān)控系統(tǒng)

資源監(jiān)控系統(tǒng)為RFTS算法后期的區(qū)域劃分和基于人工螢火蟲優(yōu)化的任務(wù)調(diào)度收集重要的系統(tǒng)實時信息。該系統(tǒng)每隔10 s對每個資源節(jié)點統(tǒng)計一次實時資源性能數(shù)據(jù)。在計算資源中最具代表性的資源性能數(shù)據(jù)包括CPU核數(shù)、 CPU 使用率、內(nèi)存使用率和總內(nèi)存大小，這4個指標(biāo)反映了節(jié)點當(dāng)前的負(fù)載能力。因此,本文用這4個指標(biāo)構(gòu)成的綜合值來代表該資源節(jié)點的實時性能，并通過計算得到每個資源節(jié)點的負(fù)載值，進而得到整個集群的負(fù)載值。

資源監(jiān)控系統(tǒng)的具體流程如算法1所示，當(dāng)集群負(fù)載值大于或等于β時(第1行)，對于集群中的每個資源節(jié)點TMj，根據(jù)監(jiān)控和管理Java虛擬機(JVM)管理接口的ManagementFactory管理工廠類中的getOperatingSystemMXBean方法去獲取資源節(jié)點底層的性能指標(biāo)，計算節(jié)點TMj的CPU核數(shù)Core_Numj、CPU利用率ω1j和空閑內(nèi)存量ω2j，根據(jù)式(1)和式(2)可以得出該資源節(jié)點TMj的性能指標(biāo)Metricj和負(fù)載值Loadj(第2～5行);根據(jù)式(4)得出集群的負(fù)載值，并把每個節(jié)點的負(fù)載值和集群負(fù)載值放入數(shù)據(jù)庫MongoDB中(第6、7行)，直到集群負(fù)載值小于閾值β，即集群實現(xiàn)整體的負(fù)載均衡。

算法1資源監(jiān)控系統(tǒng)實現(xiàn)算法

輸入：資源節(jié)點集合TM、每個資源節(jié)點上的任務(wù)分配情況TaskNum。

輸出：數(shù)據(jù)庫MongoDB。

1.While集群負(fù)載值大于或等于閾值βdo

2.For集群中的每一個資源節(jié)點do

3. 計算資源節(jié)點TMj的CPU核數(shù)、CPU利用率和當(dāng)前機器的空閑內(nèi)存量;

4. 更新資源節(jié)點TMj的性能指標(biāo)和負(fù)載值;

5.Endfor

6. 更新集群的整體負(fù)載值;

7. 把每個節(jié)點的負(fù)載值和集群負(fù)載值存入數(shù)據(jù)庫MongoDB中;

8. 輸出數(shù)據(jù)庫MongoDB;

9.Endwhile

下面通過一個實例來描述資源監(jiān)控系統(tǒng)的執(zhí)行過程，如圖2所示。對于整個集群而言，每隔10 s統(tǒng)計一次每個資源節(jié)點的CPU核數(shù)、CPU利用率、內(nèi)存利用率和總內(nèi)存大小，根據(jù)式(1)計算出每個節(jié)點的性能指標(biāo)Metricj，根據(jù)每個節(jié)點上當(dāng)前任務(wù)隊列中的任務(wù)數(shù)量和實時性能指標(biāo)計算每個節(jié)點的負(fù)載值Loadj和集群負(fù)載值C_Load，并把這些信息實時存入MongoDB數(shù)據(jù)庫中，重復(fù)上述過程直到集群負(fù)載值C_Load小于β。

Figure 2 Execution of the resource monitoring system圖2 資源監(jiān)控系統(tǒng)執(zhí)行過程

Figure 3 Execution of regional division圖3 區(qū)域劃分執(zhí)行過程

3.4 區(qū)域劃分算法

本文將整個集群劃分為過負(fù)載、輕負(fù)載、近飽和和差飽和4個區(qū)域，依次記為UPGroup、LPGroup、NSGroup和DSGroup。

通過定義如式(5)所示的偏移量offset來劃分集群區(qū)域。h_threshold為啟發(fā)式集群高域值，且h_threshold≥0;l_threshold為低閾值，且l_threshold≤0。如果offsetj≥h_threshold,則節(jié)點TMj屬于過負(fù)載區(qū)域UPGroup;如果0≤offsetj

offsetj=δj-β,j=1,2,…,m

(5)

其中資源節(jié)點TMj負(fù)載偏差值δj如式(6)所示：

(6)

區(qū)域劃分算法的偽代碼如算法2所示。當(dāng)滿足過負(fù)載區(qū)域UPGroup不為空且差飽和區(qū)域DSGroup不為空，或者集群負(fù)載值C_Load<β這2個條件中的一條時(第1行)開始區(qū)域劃分算法，每隔10 s從MongoDB數(shù)據(jù)庫中提取當(dāng)前資源中每個節(jié)點的負(fù)載值loadj，計算得出當(dāng)前整個集群的平均負(fù)載，根據(jù)式(5)計算得到每個資源節(jié)點的當(dāng)前偏移量offsetj(第3、4行)，并結(jié)合啟發(fā)式集群高低閾值h_threshold和l_threshold計算出該節(jié)點的負(fù)載承受能力，根據(jù)該節(jié)點負(fù)載承受能力的高低把該節(jié)點分配到對應(yīng)所屬區(qū)域(第5～16行)。如果區(qū)域劃分算法結(jié)束時C_Load小于β，則算法結(jié)束，否則，進入到基于人工螢火蟲優(yōu)化的任務(wù)調(diào)度算法重新分配任務(wù)調(diào)度集合。

算法2區(qū)域劃分算法

輸入：數(shù)據(jù)庫MongoDB。

輸出：過負(fù)載區(qū)域UPGroup、輕負(fù)載區(qū)域LPGroup、近飽和區(qū)域NSGroup和差飽和區(qū)域DSGroup。

1.While(UPGroup≠?∧DSGroup≠?)∨(C_Load<β)do

2.ForMongoDB數(shù)據(jù)庫中的每一個資源節(jié)點

3. 計算每個資源節(jié)點TMj的負(fù)載偏差值δj;

4. 計算每個資源節(jié)點TMj的偏移量offsetj;

5.Casewhenoffsetj≥h_thresholdthen

6. 把資源節(jié)點TMj放入過負(fù)載區(qū)域UPGroup;

7.When0≤offsetj

8. 把資源節(jié)點TMj放入輕負(fù)載區(qū)域LPGroup;

9.Whenoffsetj

10. 把資源節(jié)點TMj放入差飽和區(qū)域DSGroup;

11.Whenl_threshold≤offsetj<0then

12. 把資源節(jié)點TMj放入近飽和區(qū)域NSGroup;

13.EndWhile

14.輸出劃分好的過負(fù)載區(qū)域UPGroup、輕負(fù)載區(qū)域LPGroup、近飽和區(qū)域NSGroup和差飽和區(qū)域DSGroup.

下面通過一個實例來描述區(qū)域劃分算法的具體執(zhí)行過程，如圖3所示。假定當(dāng)前集群由6個TaskManager組成，根據(jù)3.3節(jié)提出的資源監(jiān)控系統(tǒng)，每個TaskManager的實時負(fù)載值都存儲在MongoDB數(shù)據(jù)庫中，每隔10 s從該數(shù)據(jù)庫中重新提取當(dāng)前資源中每個節(jié)點的負(fù)載值Loadj和集群的負(fù)載值C_Load，并計算得到每個節(jié)點相對于集群負(fù)載平均值的偏移量。該實例中,TaskManager1～TaskManager6的偏移量分別是0.91,-0.09,-0.12,-0.58,-0.31和0.87，集群的高閾值為0.83,低閾值為-0.27，根據(jù)區(qū)域劃分算法，分配結(jié)果如圖3的右圖所示，TaskManager1和TaskManager6被分配到UPGroup，TaskManager2和TaskManager3被分配到NSGroup，TaskManager4和TaskManager5被分配到DSGroup。

3.5 基于人工螢火蟲優(yōu)化的任務(wù)調(diào)度算法

人工螢火蟲優(yōu)化GSO算法是2005年由印度學(xué)者Krishnanand和Ghose提出的一種新型的全局智能優(yōu)化算法。該算法定義了螢火蟲的解空間，每個螢火蟲都有決策域，即自己的視線范圍。每只螢火蟲的亮度與其所在位置的目標(biāo)函數(shù)值有關(guān)，螢火蟲位置的目標(biāo)函數(shù)值越高，其亮度越大；相反，則亮度越小。根據(jù)螢火蟲的自然生活規(guī)律，它將在決策域中找到下一次運動方向。在決策域中，區(qū)域越亮，對螢火蟲的吸引力越強。螢火蟲的飛行方向會根據(jù)鄰域改變。另外，決策域的大小受鄰域中個體數(shù)目的影響，當(dāng)鄰域密度減小時，螢火蟲決策域半徑增大，為了發(fā)現(xiàn)更多的鄰居，鄰域密度越大，其決策域半徑越小。最終，大部分螢火蟲會在一個區(qū)域內(nèi)凝結(jié)，即達到極值點。

本節(jié)提出的基于人工螢火蟲優(yōu)化的任務(wù)調(diào)度算法的原理是，基于3.4節(jié)的區(qū)域劃分算法把過負(fù)載區(qū)域UPGroup中的節(jié)點TMj上的任務(wù)Ti按照基于人工螢火蟲優(yōu)化的任務(wù)調(diào)度策略調(diào)度到差飽和區(qū)域DSGroup中的節(jié)點TMp上，UPGroup中的節(jié)點按照負(fù)載值的大小降序排序，依次調(diào)度到DSGroup中，直到集群負(fù)載值C_Load小于閾值β。在該算法中任務(wù)Ti被定義為螢火蟲，DSGroup區(qū)域中的節(jié)點TMp被定義為螢火蟲的目標(biāo)區(qū)域，任務(wù)的目標(biāo)函數(shù)由UPGroup中節(jié)點TMj的任務(wù)數(shù)量、節(jié)點TMj中的任務(wù)Ti的優(yōu)先級、目標(biāo)節(jié)點TMp的負(fù)載值和目標(biāo)節(jié)點TMp上高優(yōu)先級任務(wù)的數(shù)量共同決定。下面先對本文提出的基于人工螢火蟲優(yōu)化的任務(wù)調(diào)度算法進行基本定義與概念說明：

令N={n1,n2,…,nn}為螢火蟲集合，初始化每個螢火蟲的熒光素為l0，決策域為r0。螢火蟲ni在時刻t的熒光素值如式(7)所示:

li(t)=(1-ρ)li(t-1)+γf(xi(t)),

i=1,2,…,n

(7)

其中，ρ代表螢火蟲中熒光素的消失率，γ代表熒光素的更新率，f(xi(t))表示螢火蟲ni在時刻t時位置xi(t)的目標(biāo)函數(shù)值。

螢火蟲ni在t時刻的鄰居集合Gi(t)如式(8)所示:

li(t)

(8)

螢火蟲ni的速度方向v的定義如式(9)所示。

v=max(pi),pi={pi1,pi2,…,piGi(t)}

(9)

其中,

(10)

為螢火蟲ni向螢火蟲nj方向的轉(zhuǎn)移概率。

螢火蟲ni在時刻t時的位置定義如式(11)所示:

(11)

其中,s為螢火蟲的步長。

φ(gt-1-|Gi(t-1)|)}}

(12)

其中,rs表示螢火蟲的感知域范圍，φ表示決策域的更新率，gt-1表示螢火蟲在t-1時刻的鄰域閾值，Gi(t-1)表示螢火蟲ni在t-1時刻的鄰域。

基于人工螢火蟲算法的任務(wù)調(diào)度算法的流程如算法3所示，首先把UPGroup中的任務(wù)按照負(fù)載值高低進行排序，把UPGroup中的任務(wù)設(shè)定為螢火蟲，DSGroup區(qū)域設(shè)定為發(fā)光區(qū)域，在發(fā)光區(qū)域中尋找最優(yōu)解并把任務(wù)調(diào)度到最優(yōu)的資源節(jié)點上。

算法3基于人工螢火蟲優(yōu)化的任務(wù)調(diào)度算法

輸入：螢火蟲集合N={n1,n2,…,nn}、迭代次數(shù)M、過負(fù)載區(qū)域UPGroup、輕負(fù)載區(qū)域LPGroup、近飽和區(qū)域NSGroup、差飽和區(qū)域DSGroup。

輸出：把UPGroup區(qū)域的任務(wù)調(diào)度給DSGroup中的節(jié)點DataNode。

1. 初始化算法中需要用到的參數(shù);

2.ForniinN

3. 初始化熒光素、初始位置和初始決策域;

4.While當(dāng)UPGroup區(qū)域和DSGroup區(qū)域都不空時，才有任務(wù)可以調(diào)度的空間do

Figure 4 Execution process of task scheduling algorithm based on GSO algorithm圖4 基于人工螢火蟲算法的任務(wù)調(diào)度算法執(zhí)行過程

5.ForniinN

7.For鄰域集合中的螢火蟲nj

8. 根據(jù)式(10)更新螢火蟲ni向螢火蟲nj方向的轉(zhuǎn)移概率;

9.If螢火蟲ni向螢火蟲nj方向的轉(zhuǎn)移概率大于此時最大值maxthen

10. 更新最大值為此時的轉(zhuǎn)移概率;

11.Endif

12.Endfor

13. 選出最大值作為螢火蟲ni的移動速度方向v;

14. 使螢火蟲ni向下一次迭代速度方向v移動并更新螢火蟲移動后的位置點;

15.Endfor

16. 更新螢火蟲在t+1時刻的決策域范圍并求出全局最優(yōu)點pbest；

17.For過負(fù)載區(qū)域UPGroup中的任務(wù)Tj

18. 把任務(wù)Tj調(diào)度到差飽和區(qū)域DSGroup中選出的最優(yōu)節(jié)點pbest;

19.Endfor

20.Endwhile

21.Endfor

22. 輸出更新過后的過負(fù)載區(qū)域UPGroup、輕負(fù)載區(qū)域LPGroup、近飽和區(qū)域NSGroup和差飽和區(qū)域DSGroup集合.

下面通過一個實例來描述基于人工螢火蟲優(yōu)化的任務(wù)調(diào)度算法的具體執(zhí)行過程，如圖4所示。

圖4a為原本的區(qū)域分配情況以及任務(wù)和資源節(jié)點的對應(yīng)調(diào)度情況，當(dāng)UPGroup和DSGroup都不為空時，把過負(fù)載區(qū)域UPGroup中的資源節(jié)點根據(jù)負(fù)載值由高到低進行排序，并把該資源節(jié)點中的任務(wù)按照優(yōu)先級由高到低進行排序，對于UPGroup中的節(jié)點對應(yīng)的分配任務(wù)模擬為螢火蟲，DSGroup定義為螢火蟲的區(qū)域移動范圍。

在本次實例中UPGroup中資源節(jié)點為TaskManager1和TaskManager6，它們的偏移量分別是0.91和0.87，TaskManager1中的任務(wù)Task1、Task2、Task3的優(yōu)先級分別是低、低和高，TaskManager1中的任務(wù)Task3的優(yōu)先級分高；對于DSGroup中的TaskManager4和TaskManager5的偏移量分別是-0.58和-0.31;NSGroup中資源節(jié)點為TaskManager2和TaskManager3，LPGroup區(qū)域為空。因為UPGroup中任務(wù)重新調(diào)度的順序為先按照資源節(jié)點中偏移量由高到低排序，對于同一資源節(jié)點中的任務(wù)再按照任務(wù)的優(yōu)先級由高到低排序，因此UPGroup中第一個被調(diào)度的任務(wù)為TaskManager1中的Task3，由于DSGroup中TaskManager4的偏移量低于TaskManager5的偏移量，因此TaskManager1中的Task3任務(wù)被調(diào)度到TaskManager4中。

此時系統(tǒng)實時性能發(fā)生改變，從MongoDB數(shù)據(jù)庫中獲取實時的資源節(jié)點負(fù)載值信息，并根據(jù)節(jié)點當(dāng)前的任務(wù)隊列數(shù)量進行重新分區(qū)。此時集群任務(wù)分配情況如圖4b所示，TaskManager1偏移量更新為0.64,處于LPGroup區(qū)域，下一個被調(diào)度的任務(wù)為UPGroup中的TaskManager6的Task10，此時DSGroup中的TaskManager4和TaskManager5的偏移量分別是-0.37和-0.28，因此Task10被調(diào)度到TaskManager4中，并且資源節(jié)點的性能發(fā)生變化;此時集群重新劃分后如圖4c所示，TaskManager4被劃分到LPGroup，TaskManager6被劃分到DSGroup區(qū)域，此時UPGroup為空，即最終的任務(wù)調(diào)度分配方案。

4 實驗與分析

本文通過修改Flink 1.8.0的源碼，實現(xiàn)了基于資源反饋的負(fù)載均衡任務(wù)調(diào)度算法RFTS。本節(jié)設(shè)計并實施了一系列實驗，基于TPC-C和TPC-H數(shù)據(jù)集，從執(zhí)行時間和吞吐量2個方面對Flink默認(rèn)的調(diào)度算法、公平調(diào)度算法、遺傳算法和本文提出的RFTS算法進行對比實驗，測試了本文提出的面向Flink系統(tǒng)的任務(wù)調(diào)度優(yōu)化算法的實用性。

4.1 實驗環(huán)境配置

實驗所用環(huán)境為4個節(jié)點組成的分布式集群，包括1個主節(jié)點(Master)和3個從節(jié)點(Slave)，每臺服務(wù)器的配置信息如表1所示。

Table 1 Hardware configuration表1 硬件配置

搭建成由4臺服務(wù)器組成的Flink集群，其中的1臺Master為Flink集群中的JobManager節(jié)點，3臺Slave節(jié)點為Flink中的TaskManager節(jié)點，節(jié)點間通過千兆以太網(wǎng)連接，節(jié)點間的運行方式為Standalone模式。Flink集群的軟件及其版本如表2所示。

Table 2 Software configuration表2 軟件配置

4.2 數(shù)據(jù)集

本文實驗使用TPC-C和TPC-H數(shù)據(jù)集分別生成6個不同大小的數(shù)據(jù)集，測試程序?qū)⒃谶@12個數(shù)據(jù)集上進行測試。數(shù)據(jù)集的來源和規(guī)模如表3所示。

Table 3 Dataset size表3 數(shù)據(jù)集規(guī)模

下面分別介紹這2個數(shù)據(jù)集的數(shù)據(jù)特征:

TPC-C是聯(lián)機交易處理系統(tǒng)OLTP(On-Line Transaction Processing)的規(guī)范，TPC-C測試中使用的模型是一家大型商品批發(fā)銷售公司，在不同地區(qū)中設(shè)有多個倉庫，隨著業(yè)務(wù)的增長，公司需要添加新的倉庫，每個倉庫有10個銷售點，每個銷售點為3 000個客戶提供服務(wù)，每個銷售訂單對應(yīng)10種產(chǎn)品，大約有1%的產(chǎn)品顯示缺貨時，需要從其他區(qū)域的倉庫中調(diào)運。整個TPC-C數(shù)據(jù)集由9張表組成，包括客戶表、區(qū)域、訂單表等等，產(chǎn)生的交易事務(wù)主要有5種，分別是新訂單、支付操作、發(fā)貨、訂單狀態(tài)查詢和庫存狀態(tài)查詢，該數(shù)據(jù)集可以通過命令指定生成數(shù)據(jù)集的大小。

TPC-H是商品零售業(yè)決策支持系統(tǒng)的測試基準(zhǔn)，測試系統(tǒng)中復(fù)雜查詢的執(zhí)行時間。它包含8個基本表，數(shù)據(jù)量可以設(shè)置為1 GB～3 TB不等，其基準(zhǔn)測試包含22個查詢，查詢語句嚴(yán)格遵守SQL-92語法，并且不允許修改，主要指標(biāo)為每個請求的響應(yīng)時間，即提交任務(wù)后返回結(jié)果所花費的總時間。TPC-H中數(shù)據(jù)量的大小對查詢速度的影響很大，使用SF描述數(shù)據(jù)量，1SF對應(yīng)1 GB單位，并且人工設(shè)定的數(shù)據(jù)量只是8個表中的總數(shù)據(jù)量并不包含索引和臨時表等空間占用情況，因此設(shè)定數(shù)據(jù)時需要預(yù)留更多的空間。對于同樣規(guī)模大小的數(shù)據(jù)集，TPC-H產(chǎn)生的數(shù)據(jù)類型比TPC-C產(chǎn)生的數(shù)據(jù)類型更多樣，關(guān)系更復(fù)雜。

4.3 算法本身處理時間

本文分別在TPC-C和TPC-H不同規(guī)模的數(shù)據(jù)集上測試RFTS算法的處理時間，結(jié)果如表4所示，TPC-C代表簡單場景下的測試，TPC-H代表復(fù)雜場景下的測試。隨著數(shù)據(jù)集規(guī)模的增大，算法處理時間占任務(wù)整體執(zhí)行時間的比例也增大了，這是因為處理任務(wù)的數(shù)據(jù)規(guī)模越大，出現(xiàn)負(fù)載不均的情況越多，需要遷移的任務(wù)急劇增多，因此調(diào)度算法所占整體執(zhí)行時間的本身的比例也就越大。從表4可以看出，無論對于簡單數(shù)據(jù)還是復(fù)雜數(shù)據(jù)而言，調(diào)度算法所占整體執(zhí)行時間的比例都小于1%。

其中,Dataset7～Dataset12對應(yīng)的數(shù)據(jù)是基于TPC-H數(shù)據(jù)集的測試結(jié)果，Dataset1～Dataset6對應(yīng)的數(shù)據(jù)是基于TPC-C數(shù)據(jù)集的測試結(jié)果。通過對比Dataset1&Dataset7,Dataset2&Dataset8,Dataset3&Dataset9,Dataset4&Dataset10,Dataset5&Dataset11,Dataset6&Dataset12，可以發(fā)現(xiàn)基于TPC-H數(shù)據(jù)集的實驗中任務(wù)整體執(zhí)行時間和調(diào)度算法本身占用的時間都比基于TPC-C的要大，但是算法處理時間占任務(wù)整體執(zhí)行時間的比例反而更小了。這是因為對于復(fù)雜數(shù)據(jù)場景，負(fù)載不均的情況更常見，RFTS算法的調(diào)度算法本身增加的時間對于優(yōu)化的系統(tǒng)執(zhí)行效率來說影響更小，即優(yōu)化的效果更好。實驗結(jié)果表明，無論是TPC-C對應(yīng)的簡單環(huán)境還是TPC-H對應(yīng)的復(fù)雜場景，算法本身處理時間占任務(wù)整體執(zhí)行時間的比例都很小。

Table 4 Proportion of processing time of the RFTS algorithm表4 RFTS算法的處理時間占比

4.4 執(zhí)行時間對比分析

執(zhí)行時間是指從任務(wù)提交到完成所需要的總時間。執(zhí)行時間越少，代表系統(tǒng)處理任務(wù)的計算能力越強。本節(jié)對RFTS算法、公平調(diào)度算法(Fair)、遺傳算法(Genetic)和Flink默認(rèn)的輪詢調(diào)度算法(Default)在任務(wù)整體執(zhí)行時間上進行對比實驗，分別基于TPC-C和TPC-H的6個數(shù)據(jù)集進行測試，作業(yè)并行度都設(shè)置為12，測試用例為WordCount。圖5和圖6分別為基于TPC-C和TPC-H數(shù)據(jù)集的執(zhí)行時間對比圖。

Figure 5 Comparison of execution time based on TPC-C dataset圖5 基于TPC-C數(shù)據(jù)集的執(zhí)行時間對比

Figure 6 Comparison of execution time based on TPC-H dataset圖6 基于TPC-H數(shù)據(jù)集的執(zhí)行時間對比

從圖5和圖6可以看出，采用Fair調(diào)度算法與默認(rèn)的輪詢調(diào)度算法的執(zhí)行效率相差不大；采用遺傳算法Genetic的執(zhí)行效率要優(yōu)于采用Fair調(diào)度和默認(rèn)的輪詢調(diào)度算法的，優(yōu)化效果約為3.1%，在數(shù)據(jù)集為11 GB時優(yōu)化效果最好；而采用本文提出的RFTS算法比采用Flink默認(rèn)的輪詢調(diào)度算法、Fair調(diào)度算法和Genetic算法的執(zhí)行時間都要短，且隨著數(shù)據(jù)集規(guī)模的增大,執(zhí)行時間的優(yōu)化效果越好，且在數(shù)據(jù)集規(guī)模大小相同的情況下，圖6對應(yīng)的基于TPC-H數(shù)據(jù)集的執(zhí)行時間優(yōu)化效果優(yōu)于圖5對應(yīng)的基于TPC-C數(shù)據(jù)集的執(zhí)行時間優(yōu)化效果。原因同上，這說明RFTS算法對于復(fù)雜的、大規(guī)模場景執(zhí)行時間優(yōu)化效果更好。但是，無論是在TPC-C對應(yīng)的簡單場景還是在TPC-H對應(yīng)的復(fù)雜場景下，采用RFTS算法后的整體任務(wù)執(zhí)行時間都要少于采用Flink默認(rèn)的輪詢?nèi)蝿?wù)調(diào)度算法的。在多種數(shù)據(jù)集的測試下最終求出采用RFTS算法時整體任務(wù)執(zhí)行時間的平均優(yōu)化效率為6.3%。

4.5 吞吐量對比分析

吞吐量(throughput)是指系統(tǒng)單位時間內(nèi)能夠處理的數(shù)據(jù)量大小，代表了系統(tǒng)的負(fù)載能力。圖7為RFTS算法、公平調(diào)度算法(Fair)、遺傳算法(Genetic)和 Flink默認(rèn)的調(diào)度算法(Default)在不同并行度下的吞吐量對比分析圖。分析圖7可知，采用Fair算法時的吞吐量比采用Flink默認(rèn)的輪詢算法的低；采用Genetic算法時的吞吐量優(yōu)于采用Flink默認(rèn)的輪詢算法和Fair算法的，在多種數(shù)據(jù)集測試下最終求出的采用Genetic算法的吞吐量平均優(yōu)化效率為3.9%；而采用RFTS算法相比采用Flink默認(rèn)的輪詢算法、Fair算法和Genetic算法的吞吐量更大，并且隨著并行度的增大，增加了同一時間內(nèi)處理任務(wù)的機器數(shù)量，系統(tǒng)承受負(fù)載的能力增大，即系統(tǒng)單位時間內(nèi)可以處理更多的數(shù)據(jù)。相比而言，使用RFTS算法吞吐量優(yōu)化效果更好，這是因為使用RFTS算法會實時監(jiān)控系統(tǒng)性能，并根據(jù)每個資源節(jié)點的負(fù)載承受能力，去調(diào)整任務(wù)隊列中的任務(wù)數(shù)量，時刻保證集群負(fù)載均衡，大大提升了資源利用率以及單位時間內(nèi)可以處理的數(shù)據(jù)量。在多種數(shù)據(jù)集測試下最終求出的采用RFTS算法的吞吐量平均優(yōu)化效率為11.7%。

Figure 7 Comparison of throughput at different degrees of parallelism圖7 不同并行度下吞吐量對比

5 結(jié)束語

Flink作為現(xiàn)在主流的大數(shù)據(jù)計算引擎，在實時數(shù)據(jù)處理和離線數(shù)據(jù)處理上都表現(xiàn)出了良好的效果，然而Flink計算引擎中的任務(wù)調(diào)度還有許多待優(yōu)化的空間，因此本文提出了基于資源反饋的負(fù)載均衡任務(wù)調(diào)度算法RFTS。實驗結(jié)果表明，本文提出的RFTS算法能夠有效減少Flink計算引擎中任務(wù)的整體執(zhí)行時間，增加吞吐量。

未來希望本文提出的RFTS算法可以應(yīng)用于其它的大數(shù)據(jù)計算引擎中，并取得性能提升。