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

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

1 引言

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

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

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

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

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

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

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

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

2 相關工作

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

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

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

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

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

3 基于資源反饋的負載均衡任務調度算法RFTS

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

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

3.1 定義與說明

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

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

定義2(資源節(jié)點性能指標) 資源節(jié)點的性能指標如式(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的內存利用率，ω3j表示資源節(jié)點TMj的總內存大小。因此，資源節(jié)點性能指標Metricj由CPU核數(shù)、CPU利用率和空閑內存大小決定，最后作歸一化處理。

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

(2)

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

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

(3)

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

(4)

3.2 主要思想

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

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

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

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

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

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

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

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

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

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

1.While集群負載值大于或等于閾值βdo

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

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

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

5.Endfor

6. 更新集群的整體負載值;

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

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

9.Endwhile

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

本文將整個集群劃分為過負載、輕負載、近飽和和差飽和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屬于過負載區(qū)域UPGroup;如果0≤offsetj

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

(5)

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

(6)

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

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

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

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

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

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

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

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

5.Casewhenoffsetj≥h_thresholdthen

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

7.When0≤offsetj

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

9.Whenoffsetj

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

11.Whenl_threshold≤offsetj<0then

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

13.EndWhile

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

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

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

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

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

令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)的目標函數(shù)值。

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

li(t)

(8)

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

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

(9)

其中,

(10)

為螢火蟲ni向螢火蟲nj方向的轉移概率。

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

(11)

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

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

(12)

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

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

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

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

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

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

2.ForniinN

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

4.While當UPGroup區(qū)域和DSGroup區(qū)域都不空時，才有任務可以調度的空間do

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

5.ForniinN

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

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

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

10. 更新最大值為此時的轉移概率;

11.Endif

12.Endfor

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

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

15.Endfor

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

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

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

19.Endfor

20.Endwhile

21.Endfor

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

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

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

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

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

4 實驗與分析

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

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

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

Table 1 Hardware configuration表1 硬件配置

搭建成由4臺服務器組成的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ù)集，測試程序將在這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ū)中設有多個倉庫，隨著業(yè)務的增長，公司需要添加新的倉庫，每個倉庫有10個銷售點，每個銷售點為3 000個客戶提供服務，每個銷售訂單對應10種產品，大約有1%的產品顯示缺貨時，需要從其他區(qū)域的倉庫中調運。整個TPC-C數(shù)據(jù)集由9張表組成，包括客戶表、區(qū)域、訂單表等等，產生的交易事務主要有5種，分別是新訂單、支付操作、發(fā)貨、訂單狀態(tài)查詢和庫存狀態(tài)查詢，該數(shù)據(jù)集可以通過命令指定生成數(shù)據(jù)集的大小。

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

4.3 算法本身處理時間

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

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

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

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

執(zhí)行時間是指從任務提交到完成所需要的總時間。執(zhí)行時間越少，代表系統(tǒng)處理任務的計算能力越強。本節(jié)對RFTS算法、公平調度算法(Fair)、遺傳算法(Genetic)和Flink默認的輪詢調度算法(Default)在任務整體執(zhí)行時間上進行對比實驗，分別基于TPC-C和TPC-H的6個數(shù)據(jù)集進行測試，作業(yè)并行度都設置為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調度算法與默認的輪詢調度算法的執(zhí)行效率相差不大；采用遺傳算法Genetic的執(zhí)行效率要優(yōu)于采用Fair調度和默認的輪詢調度算法的，優(yōu)化效果約為3.1%，在數(shù)據(jù)集為11 GB時優(yōu)化效果最好；而采用本文提出的RFTS算法比采用Flink默認的輪詢調度算法、Fair調度算法和Genetic算法的執(zhí)行時間都要短，且隨著數(shù)據(jù)集規(guī)模的增大,執(zhí)行時間的優(yōu)化效果越好，且在數(shù)據(jù)集規(guī)模大小相同的情況下，圖6對應的基于TPC-H數(shù)據(jù)集的執(zhí)行時間優(yōu)化效果優(yōu)于圖5對應的基于TPC-C數(shù)據(jù)集的執(zhí)行時間優(yōu)化效果。原因同上，這說明RFTS算法對于復雜的、大規(guī)模場景執(zhí)行時間優(yōu)化效果更好。但是，無論是在TPC-C對應的簡單場景還是在TPC-H對應的復雜場景下，采用RFTS算法后的整體任務執(zhí)行時間都要少于采用Flink默認的輪詢任務調度算法的。在多種數(shù)據(jù)集的測試下最終求出采用RFTS算法時整體任務執(zhí)行時間的平均優(yōu)化效率為6.3%。

4.5 吞吐量對比分析

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

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

5 結束語

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

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