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

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

?

異構環(huán)境下基于雙重預取的Hadoop調度算法

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

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

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

Hadoop；異構環(huán)境；調度算法；雙重預取

0 引言

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

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

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

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

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

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

圖1 MapReduce框架

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

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

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

2 改進的算法

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

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

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

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

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

具體步驟如下：

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

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

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

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

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

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

(1)

(2)

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

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

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

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

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

該預取算法偽代碼描述如下：

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任務分配給node節(jié)點

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

nodes begin data perfecting;

endif

end for

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

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

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

(3)

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

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

(4)

(5)

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

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

whileRate

Rate=MapNumsfinished/MapNumstotal

endwhile

while!(allmapshasfinashed)do

foreachnodeinNodesWithReduceTaskdo

iftaskneedsmoredatado

prefectchingdata

if預取沖突then

WaitingTime=rate*TIME//等待

endif

endif

endfor

endwhile

3 實驗

本實驗采用虛擬機的方式搭建異構環(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)境配置

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

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

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

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

4 總結

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

[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)延遲調度機制[J]．小型微型計算機系統(tǒng)，2015，3(3)：445-449．

[7] 金嘉暉，羅軍舟，宋愛波，等．基于數(shù)據(jù)中心負載分析的自適應延遲調度算法[J]．通信學報，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ù)局部性改進調度算法[J]．計算機科學，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。

國家自然科學基金項目(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