異構(gòu)環(huán)境下基于雙重預(yù)取的Hadoop調(diào)度算法

孫玉強，陸 勇，王文聞，李媛媛，顧玉宛

(常州大學(xué) 信息科學(xué)與工程學(xué)院,江蘇 常州 213000)

?

異構(gòu)環(huán)境下基于雙重預(yù)取的Hadoop調(diào)度算法

孫玉強，陸 勇，王文聞，李媛媛，顧玉宛

(常州大學(xué) 信息科學(xué)與工程學(xué)院,江蘇 常州 213000)

Hadoop處理海量數(shù)據(jù)時，無論是Map任務(wù)還是Reduce任務(wù)都需要耗費大量的時間傳輸數(shù)據(jù)，故提出一種基于雙重預(yù)取的調(diào)度算法；該算法通過估算節(jié)點上任務(wù)執(zhí)行的進度來預(yù)測Map任務(wù)的執(zhí)行節(jié)點，然后通知節(jié)點提前預(yù)取所需的數(shù)據(jù)，并且在Map任務(wù)完成的數(shù)量達到預(yù)定值時，開始為Reduce任務(wù)預(yù)取部分數(shù)據(jù)；由于在異構(gòu)的環(huán)境下集群中節(jié)點的性能各不相同，為此采取了改進的預(yù)測模型，以提高任務(wù)進度判斷的準(zhǔn)確性；實驗證明，本算法在作業(yè)響應(yīng)時間等方面優(yōu)于現(xiàn)有的調(diào)度算法。

Hadoop；異構(gòu)環(huán)境；調(diào)度算法；雙重預(yù)取

0 引言

近年來，隨著計算機的普及和信息技術(shù)的快速發(fā)展，互聯(lián)網(wǎng)的使用也越來越廣泛。與此同時，互聯(lián)網(wǎng)上的數(shù)據(jù)也呈現(xiàn)出爆炸式的增長。例如，紐約證券交易所每天產(chǎn)生1TB的交易數(shù)據(jù)。FaceBook存儲著超過1億張照片，約1PB存儲容量。據(jù)預(yù)測，2015 年全球產(chǎn)生的數(shù)據(jù)量將達到近10 ZB，而2020年全球產(chǎn)生的數(shù)據(jù)量將達到40 ZB。我們已經(jīng)生活在一個大數(shù)據(jù)的時代，越來越多的公司需要面對大數(shù)據(jù)的處理問題。

傳統(tǒng)的的解決方案存在著存儲量小、穩(wěn)定性差、耗時過長等缺點。當(dāng)前廣泛使用的大數(shù)據(jù)處理模型是由Google公司設(shè)計的MapReduce[1]編程模型。MapReduce作為并行分布式數(shù)據(jù)處理框架獲得了極大的成功。MapReduce把海量的數(shù)據(jù)計算劃分為許多小的任務(wù)，并且讓它們在不同的節(jié)點上并行執(zhí)行。它隱藏了底層處理的細節(jié)，為開發(fā)分布式應(yīng)用提供了簡單的編程接口。由于其具有良好的可擴展性、容錯性、可用性等特點，使得其成為近年來數(shù)據(jù)處理領(lǐng)域的研究熱點。Hadoop[2]平臺作為MapReduce的開源實現(xiàn)，由于其良好的性能，已經(jīng)被Yahoo!、Amazon等公司采用。

MapReduce的處理過程主要分為兩步。首先是執(zhí)行Map任務(wù)，處理輸入數(shù)據(jù)并生成中間結(jié)果，將其存儲在本地節(jié)點。然后Reduce任務(wù)遠程讀取這些中間結(jié)果，經(jīng)過運算產(chǎn)生最終的輸出結(jié)果。任務(wù)的執(zhí)行需要調(diào)度器將用戶作業(yè)隊列中的任務(wù)分配給相應(yīng)的節(jié)點，調(diào)度算法的優(yōu)劣對于系統(tǒng)的性能起著至關(guān)重要的作用。為此，大量的學(xué)者對Hadoop的調(diào)度算法做了相關(guān)的研究。

常見的Hadoop調(diào)度算法有FIFO[3]調(diào)度，HOD[4]調(diào)度，公平調(diào)度[5]等算法。陶永才等人提出基于動態(tài)負載均衡的Hadoop動態(tài)延遲調(diào)度機制[6]，金嘉暉等人提出基于數(shù)據(jù)中心負載分析的自適應(yīng)延遲調(diào)度算法[7]，兩種方法均提高了作業(yè)的響應(yīng)時間，但都沒有考慮節(jié)點的異構(gòu)性。M Zaharia等人針對異構(gòu)環(huán)境提出推測任務(wù)剩余時間的LATE[8]算法，李麗英等人基于LATE算法提出數(shù)據(jù)局部性改進的調(diào)度算法[9]。但上述算法都缺乏對數(shù)據(jù)預(yù)取方面的考慮。本文提出異構(gòu)環(huán)境下基于雙重預(yù)取的Hadoop調(diào)度算法，以提高作業(yè)的執(zhí)行效率。

1 MapReduce框架分析及問題的提出

如圖1所示，MapReduce中任務(wù)的執(zhí)行分為Map和Reduce兩個階段。在用戶提交了一個作業(yè)后，輸入數(shù)據(jù)被劃分為多個數(shù)據(jù)塊(默認為64 M)，每個數(shù)據(jù)塊對應(yīng)一個Map任務(wù)。Map任務(wù)對輸入進行處理生成對作為中間結(jié)果。當(dāng)所有的Map任務(wù)執(zhí)行完成之后，Map輸出的中間結(jié)果交給Reduce任務(wù)執(zhí)行。Reduce任務(wù)可以分為三步。首先執(zhí)行Reduce任務(wù)的節(jié)點讀取中間結(jié)果，然后根據(jù)key進行排序，最后調(diào)用用戶編寫的Reduce函數(shù)執(zhí)行并輸出最終結(jié)果。

圖1 MapReduce框架

Hadoop集群中有一個JobTracker負責(zé)調(diào)度作業(yè)運行。當(dāng)JobTracker收到客戶端提交的作業(yè)后，就把它放在一個隊列中。調(diào)度器為其創(chuàng)建相應(yīng)的Map和Reduce任務(wù)。其余的節(jié)點作為TaskTracker負責(zé)具體任務(wù)的執(zhí)行并且通過心跳信號(heartbeat)與JobTracker進行通信。當(dāng)某個TaskTracker有空閑資源時，就會向JobTracker請求新的任務(wù)，此時調(diào)度器會給這個節(jié)點分配一個Map或者Reduce任務(wù)。

通過分析MapReduce的執(zhí)行流程可以發(fā)現(xiàn)，Map任務(wù)執(zhí)行前如果沒有取到相應(yīng)的數(shù)據(jù)，會先遠程讀取所需要的數(shù)據(jù)。如果數(shù)據(jù)量非常大，那么Map任務(wù)的數(shù)量也較多，并且它們分布在集群中不同的節(jié)點執(zhí)行，這將導(dǎo)致大量的數(shù)據(jù)傳輸開銷。另外，Reduce階段要等到所有的Map任務(wù)執(zhí)行結(jié)束后才開始，并且同樣需要傳輸大量數(shù)據(jù)。這些都會耗費大量的時間。

針對以上問題，本文將采用改進的調(diào)度算法，通過雙重數(shù)據(jù)預(yù)取的方式來減少任務(wù)執(zhí)行時讀取數(shù)據(jù)的時間，提升Hadoop的執(zhí)行效率。

2 改進的算法

本文提出的改進算法從兩方面進行數(shù)據(jù)的預(yù)取。

1)在節(jié)點分配Map任務(wù)之前，通過預(yù)測模型找出最快即將完成任務(wù)的節(jié)點，通知相應(yīng)的節(jié)點進行下次Map任務(wù)所需數(shù)據(jù)的預(yù)取。

2)在所有Map任務(wù)完成之前，即將進行Reduce任務(wù)的節(jié)點對Map任務(wù)已經(jīng)生成的中間數(shù)據(jù)進行預(yù)取。

2.1 Map任務(wù)的數(shù)據(jù)預(yù)取

在MapReduce中，當(dāng)一個節(jié)點被分配Map任務(wù)后，首先要獲取任務(wù)所需要的輸入數(shù)據(jù)。在最好的情況下，運行的任務(wù)和需要的數(shù)據(jù)在同一個節(jié)點上，則稱其滿足數(shù)據(jù)本地性，否則需要從遠程節(jié)點進行讀取。為了減少數(shù)據(jù)讀取的時間，可采用數(shù)據(jù)預(yù)取的方式，使得當(dāng)前任務(wù)的執(zhí)行與下次任務(wù)數(shù)據(jù)的傳輸并行執(zhí)行。

具體步驟如下：

1)TaskTracker節(jié)點檢查其正在執(zhí)行的任務(wù)，推測出任務(wù)結(jié)束還需要的剩余時間，并通知JobTracker節(jié)點。

2)JobTracker節(jié)點生成一個預(yù)調(diào)度節(jié)點隊列PreNodes，并且設(shè)定一個閾值MapThreshold，把剩余時間低于閾值的節(jié)點加入隊列。時間越短的在隊列中的位置越靠前。

3)JobTracker從任務(wù)隊列中取出即將調(diào)度的任務(wù)，預(yù)先將其分配給隊列中最前面的節(jié)點，并且通知其預(yù)取相應(yīng)的數(shù)據(jù)。

4)接收到預(yù)取數(shù)據(jù)任務(wù)的節(jié)點開始預(yù)取下一次Map任務(wù)的數(shù)據(jù)。

首先，為了讓節(jié)點推測出任務(wù)執(zhí)行需要的剩余時間，就需要一個推測模型。Hadoop根據(jù)任務(wù)的執(zhí)行進度判斷任務(wù)執(zhí)行的快慢。但是在異構(gòu)的環(huán)境下，每個節(jié)點的CPU、內(nèi)存以及I/O性能都不一樣，因此上述方法并不適用于判斷任務(wù)的剩余完成時間。目前有一種針對異構(gòu)環(huán)境的LATE算法，其核心思想是通過執(zhí)行進度判斷任務(wù)的速率進而推算出任務(wù)的剩余完成時間，這正符合我們的目的。

LATE算法的計算公式如下所示：

(1)

(2)

其中:ProgressScore是任務(wù)執(zhí)行的進度，量化為百分比。T是任務(wù)已經(jīng)運行的時間，TimeRemain是任務(wù)完成需要的剩余時間。

ProgressScore的值基于任務(wù)執(zhí)行的階段。在此處我們僅僅關(guān)注Map任務(wù)，Map任務(wù)的執(zhí)行被劃分為兩個子階段：1)Map函數(shù)執(zhí)行，占2/3；2)sort和partition階段，占1/3。但是由于在異構(gòu)環(huán)境下節(jié)點差異較大，任務(wù)執(zhí)行的各階段比例也會有所不同，所以采用固定的劃分方式并不合理。為此，本算法在TaskTracker執(zhí)行Map任務(wù)時，把各階段的比例信息保存在磁盤中。下次執(zhí)行任務(wù)前，首先讀取應(yīng)用的歷史執(zhí)行信息，動態(tài)確定各階段的比例，使得每個節(jié)點都能得到更準(zhǔn)確的進度值。ProgressScore與運行時間T的比值則為當(dāng)前任務(wù)的速率ProgressRate，用任務(wù)剩余進度除以速率則可以估算出任務(wù)完成需要的時間。

為了實現(xiàn)預(yù)取數(shù)據(jù)的保存，TaskTracker節(jié)點必須做相應(yīng)的改進。鑒于目前節(jié)點的內(nèi)存都比較大，可直接把數(shù)據(jù)預(yù)取到內(nèi)存中，加快Map任務(wù)的執(zhí)行速度。為此，把節(jié)點的內(nèi)存劃分為兩部分。一部分存放當(dāng)前任務(wù)的數(shù)據(jù)，另一部分用于存放下次Map任務(wù)的數(shù)據(jù)。當(dāng)前任務(wù)結(jié)束后，其所使用的數(shù)據(jù)空間將用于下次任務(wù)預(yù)取數(shù)據(jù)的存儲，而當(dāng)前的預(yù)取數(shù)據(jù)將成為下次任務(wù)的輸入數(shù)據(jù)。這兩部分空間輪換使用，從而達到數(shù)據(jù)預(yù)取的目的。

如圖2所示，當(dāng)JobTracker節(jié)點預(yù)判到map6這個任務(wù)即將在節(jié)點1執(zhí)行時，就會給節(jié)點1下達預(yù)取指令，于是節(jié)點1從節(jié)點2開始傳輸map6所需的輸入數(shù)據(jù)。

圖2 數(shù)據(jù)預(yù)取模型

該預(yù)取算法偽代碼描述如下：

for each map in TaskTracker do

TimeRemain = (1-ProgressScore)/ ProgressRate

endfor

sort(nodes)

if nodes.TimeRemain < MapThreshold then

queue.offer(nodes)

end if

for maps in job queue do

nodes ← map //把map任務(wù)分配給node節(jié)點

if nodes沒有map任務(wù)的數(shù)據(jù)then

nodes begin data perfecting;

endif

end for

2.2 Reduce任務(wù)的數(shù)據(jù)預(yù)取

Reduce任務(wù)必須等到所有的Map任務(wù)執(zhí)行完成之后才開始執(zhí)行，第一步是從相應(yīng)的節(jié)點將Map生成的中間結(jié)果進行拷貝，一樣需要數(shù)據(jù)傳輸?shù)拈_銷。為此，同樣對Reduce任務(wù)進行數(shù)據(jù)的預(yù)取。

Redecue任務(wù)的數(shù)據(jù)預(yù)取與Map任務(wù)的預(yù)取相似，但不同的是由于Reduce任務(wù)的輸入數(shù)據(jù)來源于Map任務(wù)的輸出，所以其數(shù)據(jù)預(yù)取不能依靠當(dāng)前節(jié)點Reduce任務(wù)的剩余時間。本算法采取的策略是，利用式(3)計算已完成任務(wù)的Map數(shù)量與Map任務(wù)的總數(shù)量之比。當(dāng)比值達到一個設(shè)定的閾值ReduceThreshold之后，即開始為Reduce任務(wù)進行預(yù)取。

(3)

其中:MapNumsfinished為已完成的Map任務(wù)數(shù)，MapNumstotal為Map任務(wù)總數(shù)量。

在為Reduce任務(wù)預(yù)取中間結(jié)果的時候，可能會出現(xiàn)多個節(jié)點同時從相同的輸出結(jié)果預(yù)取數(shù)據(jù)，由此造成I/O資源的競爭，降低了預(yù)取的效率。為了解決這個問題，采用沖突檢測的方式來為Reduce任務(wù)預(yù)取數(shù)據(jù)。當(dāng)某個Reduce任務(wù)預(yù)取的數(shù)據(jù)所在節(jié)點已經(jīng)有其它任務(wù)正在預(yù)取時，則先跳過這部分數(shù)據(jù)，轉(zhuǎn)而預(yù)取其他節(jié)點上的所需數(shù)據(jù)。如果所有的節(jié)點都被其他任務(wù)占用，則等待一段時間再發(fā)起預(yù)取請求。等待時間設(shè)定如下：

(4)

(5)

式(4)中Datafinished是已經(jīng)預(yù)取的數(shù)據(jù)，Datatotal是需要預(yù)取的數(shù)據(jù)總量，Ratedata則為任務(wù)的數(shù)據(jù)預(yù)取率。TIME是一個時間常數(shù)，表示集群中傳輸一個數(shù)據(jù)塊所用時間。WaitingTime則是等待時間，與數(shù)據(jù)預(yù)取率成正比。數(shù)據(jù)預(yù)取率較高的，等待時間也較長，以保證各Reduce任務(wù)均衡預(yù)取。

由于隨著時間的推移，Map任務(wù)完成的數(shù)量越來越多，所以需要循環(huán)地探測執(zhí)行Map任務(wù)的節(jié)點，獲取更多地數(shù)據(jù)。該過程算法偽代碼描述如下：

whileRate

Rate=MapNumsfinished/MapNumstotal

endwhile

while!(allmapshasfinashed)do

foreachnodeinNodesWithReduceTaskdo

iftaskneedsmoredatado

prefectchingdata

if預(yù)取沖突then

WaitingTime=rate*TIME//等待

endif

endif

endfor

endwhile

3 實驗

本實驗采用虛擬機的方式搭建異構(gòu)環(huán)境，各PC機采用 100M的局域網(wǎng)互聯(lián)。虛擬機使用VMwareworkstation10.0安裝的CentOs6.5 32位系統(tǒng)。JDK版本為1.7.0_45，Hadoop版本為0.20.1。具體集群配置如表1所示。

表1 Hadoop集群環(huán)境配置

實驗中選取的任務(wù)是Sort和WordCount。因為這些任務(wù)涉及到大量數(shù)據(jù)的傳輸，有利于比較算法之間的差異。在HDFS中數(shù)據(jù)塊設(shè)為默認的64 M，且每個數(shù)據(jù)塊保存2個副本。圖3(a)和圖3(b)分別是Map任務(wù)和Reduce任務(wù)的數(shù)據(jù)本地性比較。

圖3 任務(wù)數(shù)據(jù)本地性對比

通過圖3可以看出，相比于傳統(tǒng)的無數(shù)據(jù)預(yù)取的FIFO算法，執(zhí)行數(shù)據(jù)預(yù)取之后的Map任務(wù)和Reduce任務(wù)的數(shù)據(jù)本地性都得到極大提高。當(dāng)然，對于作業(yè)執(zhí)行效率的分析最重要的是響應(yīng)時間。對上述任務(wù)設(shè)定不同的數(shù)據(jù)量多次執(zhí)行后得到圖4的響應(yīng)時間圖。從圖中可以推算出，通過雙重的數(shù)據(jù)預(yù)取后作業(yè)的響應(yīng)時間提升了大約18%。

圖4 作業(yè)響應(yīng)時間對比

4 總結(jié)

針對Hadoop作業(yè)執(zhí)行時的數(shù)據(jù)本地性問題，本文提出基于雙重預(yù)取的調(diào)度算法。通過預(yù)測任務(wù)的執(zhí)行節(jié)點，為Map任務(wù)預(yù)取所需的數(shù)據(jù)。在Map任務(wù)完成了一定數(shù)量后，預(yù)先為Reduce任務(wù)拷貝Map已經(jīng)生成的中間結(jié)果，解決了Reduce任務(wù)的遠程調(diào)度問題。另外，算法充分考慮異構(gòu)環(huán)境下節(jié)點性能的差異，采用改進的LATE算法以便更準(zhǔn)確地判斷任務(wù)的執(zhí)行進度。實驗證明，使用數(shù)據(jù)預(yù)取有效提高了作業(yè)的響應(yīng)時間。

[1] Dean J, Ghemawat S. MapReduce: simplified data processing on large clusters[J].Commun. ACM 51 (1) (2008) 107-113.

[2] White T. Hadoop: The Definitive Guide[Z]. O’Reilly Media,Inc., 2009.

[3] HADOOP-3759: Provide ability to run memory intensive jobs without affecting other running tasks on the nodes[EB/0L]．https://issues.apache.org/jira/browse/HADOOP-3759.

[4] Apache. Hadoop On Demand[DB/OL]. http://hadoop.apache.org/common/does/r0.17.2/hod. html, 2008,20(8).

[5] Zaharia M，Borthakur D，Sarma J S，et al．Job Scheduling for Multi-user MapReduce Clusters[R]．EECS-2009—55．April 2009.

[6] 陶永才，李文潔，石 磊，等．基于負載均衡的Hadoop動態(tài)延遲調(diào)度機制[J]．小型微型計算機系統(tǒng)，2015，3(3)：445-449．

[7] 金嘉暉，羅軍舟，宋愛波，等．基于數(shù)據(jù)中心負載分析的自適應(yīng)延遲調(diào)度算法[J]．通信學(xué)報，2011，32(7)：47-56．

[8] Zaharia M, Konwinski A, et al. Improving mapreduce performance in heterogeneous environments[A].Proc of USENIX conference on Operating systems design and implementation[C].Berkeley:USENIX Association,2008:29-42.

[9] 李麗英，唐 卓，李仁發(fā)．基于LATE的Hadoop數(shù)據(jù)局部性改進調(diào)度算法[J]．計算機科學(xué)，2011(11)：67-70.

Scheduling Algorithm Based on Double Prefetching in Heterogeneous Hadoop Clusters

Sun Yuqiang， Lu Yong， Wang Wenwen， Li Yuanyuan， Gu Yuwan

(School of Information Science & Engineering, Changzhou University, Changzhou 213000,China)

When Hadoop processing huge amounts of data, both in the Map tasks and Reduce tasks requires a lot of time to transfer data. This paper presents a scheduling algorithm based on double prefetching, the algorithm predicts the node which will execute the Map task by estimating the progress of running tasks, so that the node can prefetch required data for Map tasks. Moreover, the system can also prefetch the data for Reduce tasks while Map tasks are running. Due to the performance of the cluster nodes in heterogeneous environment are not identical, the algorithm adopts an improved prediction model to improve the accuracy of the judgment of task progress. Experiments show that the algorithm is superior to the existing scheduling algorithm with less response time.

Hadoop;heterogeneous;scheduling algotithm;double prefetching

2016-02-29；

2016-04-25。

國家自然科學(xué)基金項目(11271057)；江蘇省普通高校研究生科研創(chuàng)新計劃項目(SCZ1412800004)。

孫玉強(1956-)，男，博士，教授，主要從事并行計算、軟件工程方向的研究。

顧玉宛，女，博士生，通訊聯(lián)系人，主要從事并行計算和圖像處理方向的研究。

1671-4598(2016)09-0172-04

10.16526/j.cnki.11-4762/tp.2016.09.048

TP3

A