張霄宏，孫江峰，趙文濤(1.河南理工大學(xué) 計(jì)算機(jī)科學(xué)與技術(shù)學(xué)院，河南 焦作，454003；2.中國科學(xué)院 深圳先進(jìn)技術(shù)研究院，廣東 深圳，518055；3.河南省高等學(xué)校礦山信息化重點(diǎn)學(xué)科開放實(shí)驗(yàn)室，河南 焦作，454003)

基于PUSH機(jī)制的任務(wù)調(diào)度方法

張霄宏1,2,3，孫江峰1,3，趙文濤1,3
(1.河南理工大學(xué) 計(jì)算機(jī)科學(xué)與技術(shù)學(xué)院，河南 焦作，454003；
2.中國科學(xué)院 深圳先進(jìn)技術(shù)研究院，廣東 深圳，518055；
3.河南省高等學(xué)校礦山信息化重點(diǎn)學(xué)科開放實(shí)驗(yàn)室，河南 焦作，454003)

為降低Hadoop MapReduce環(huán)境中任務(wù)的數(shù)據(jù)訪問延時(shí)進(jìn)而提高系統(tǒng)性能，提出一種基于PUSH機(jī)制的任務(wù)調(diào)度方法。該方法根據(jù)輸入數(shù)據(jù)分布，主動將任務(wù)推送到存儲其輸入數(shù)據(jù)的節(jié)點(diǎn)。當(dāng)任務(wù)在這些節(jié)點(diǎn)執(zhí)行時(shí)，可以直接從本地磁盤讀取數(shù)據(jù)，從而避免遠(yuǎn)程數(shù)據(jù)訪問延時(shí)。該方法已在hadoop-0.20.2中實(shí)現(xiàn)，并在真實(shí)集群中進(jìn)行驗(yàn)證。研究結(jié)果表明：與原有調(diào)度方式相比，該方法可將作業(yè)執(zhí)行時(shí)間平均降低8%，在最好情況下可降低14.3%。

數(shù)據(jù)局部性；性能優(yōu)化；任務(wù)調(diào)度；MapReduce

為解決海量數(shù)據(jù)處理難題，Google公司率先提出了 MapReduce模型[1]。其開源實(shí)現(xiàn) Hadoop MapReduce[2]，已成為海量數(shù)據(jù)處理領(lǐng)域的主流模型之一，并獲得了廣泛應(yīng)用[3-7]。然而，在Hadoop MapReduce環(huán)境中，當(dāng)執(zhí)行任務(wù)的節(jié)點(diǎn)與存儲其輸入數(shù)據(jù)的節(jié)點(diǎn)不是同一節(jié)點(diǎn)時(shí)，任務(wù)在執(zhí)行過程中就不得不通過遠(yuǎn)程I/O操作來訪問輸入數(shù)據(jù)，從而引起不確定的遠(yuǎn)程數(shù)據(jù)訪問延時(shí)，降低系統(tǒng)性能。且數(shù)據(jù)訪問延時(shí)越大，系統(tǒng)性能越差。為減少數(shù)據(jù)訪問延時(shí)，Hadoop缺省的調(diào)度方法總是把任務(wù)分配到離輸入數(shù)據(jù)最近的節(jié)點(diǎn)執(zhí)行。該方法采用以節(jié)點(diǎn)為中心的PULL調(diào)度機(jī)制，即只有當(dāng)節(jié)點(diǎn)請求執(zhí)行任務(wù)時(shí)才進(jìn)行調(diào)度。受輸入數(shù)據(jù)分布和資源競爭等因素的影響，該方法無法保證把所有任務(wù)都調(diào)度到存儲其輸入數(shù)據(jù)的節(jié)點(diǎn)執(zhí)行，因此，不能解決由遠(yuǎn)程數(shù)據(jù)訪問延時(shí)引起的系統(tǒng)性能問題。ZAHARIA等[8]提出利用延時(shí)調(diào)度來優(yōu)化數(shù)據(jù)局部性，但該方法調(diào)度的是MapReduce作業(yè)，而非作業(yè)中包含的任務(wù)，因此，不能從根本上解決本文提出的問題。ZHANG等[9]的方法優(yōu)先把任務(wù)保留給存儲其輸入數(shù)據(jù)的節(jié)點(diǎn)執(zhí)行，但該方法需要對各節(jié)點(diǎn)未來請求任務(wù)的情況進(jìn)行預(yù)測。WANG等[10]的方法雖然兼顧了任務(wù)的數(shù)據(jù)局部性，但由于采用短作業(yè)優(yōu)先策略，對長作業(yè)并不公平。文獻(xiàn)[7,11-14]等介紹的調(diào)度方法雖然適用于MapReduce環(huán)境，但是并不以減少數(shù)據(jù)訪問延時(shí)為目的。此外，SUN等[15]利用預(yù)取技術(shù)來隱藏部分?jǐn)?shù)據(jù)訪問延時(shí)，但是需要對任務(wù)執(zhí)行節(jié)點(diǎn)進(jìn)行預(yù)測。為避免任務(wù)在執(zhí)行過程中引起遠(yuǎn)程數(shù)據(jù)訪問延時(shí)進(jìn)而影響系統(tǒng)性能，本文作者提出基于PUSH機(jī)制的任務(wù)調(diào)度方法。該方法主動把任務(wù)推送到輸入數(shù)據(jù)所在節(jié)點(diǎn)，使任務(wù)在執(zhí)行過程中可從本地讀取數(shù)據(jù)，從而避免了遠(yuǎn)程數(shù)據(jù)訪問延時(shí)。由于該方法僅以輸入數(shù)據(jù)分布為依據(jù)進(jìn)行任務(wù)推送，即使在資源競爭激烈的環(huán)境中仍可把任務(wù)調(diào)度到輸入數(shù)據(jù)所在節(jié)點(diǎn)執(zhí)行。

1　基于PULL機(jī)制的任務(wù)調(diào)度方法

在Hadoop MapReduce環(huán)境中，作業(yè)被劃分成若干個(gè)map任務(wù)和reduce任務(wù)。map任務(wù)直接處理作業(yè)的原始輸入數(shù)據(jù)，產(chǎn)生以形式表示的中間結(jié)果。reduce任務(wù)以這些中間結(jié)果為輸入，產(chǎn)生最終的輸出。由文獻(xiàn)[16]可知，大部分的MapReduce作業(yè)都是map任務(wù)密集型作業(yè)，且其中很多作業(yè)都只有map任務(wù)，故本文僅研究降低map任務(wù)數(shù)據(jù)訪問延時(shí)的方法。

為減少數(shù)據(jù)訪問延時(shí)，Hadoop現(xiàn)有的調(diào)度方法總是把任務(wù)分配到離輸入數(shù)據(jù)最近的節(jié)點(diǎn)執(zhí)行。由于采用PULL機(jī)制，該方法只有在收到節(jié)點(diǎn)的請求時(shí)才給它分配任務(wù)，且優(yōu)先分配數(shù)據(jù)存儲在請求節(jié)點(diǎn)上的任務(wù)。只有在無此類任務(wù)的情況下，才選擇輸入數(shù)據(jù)離請求節(jié)點(diǎn)最近的任務(wù)。如果所選任務(wù)的輸入數(shù)據(jù)存儲在下一個(gè)請求節(jié)點(diǎn)上，那么該任務(wù)便錯(cuò)過了在輸入數(shù)據(jù)所在節(jié)點(diǎn)執(zhí)行的機(jī)會。

圖1　基于Pull機(jī)制的任務(wù)調(diào)度過程Fig.1　Scheduling tasks based on the pull mechanism

為便于描述，記ni表示第i個(gè)節(jié)點(diǎn)，mi表示第i個(gè)任務(wù)，di表示mi的輸入數(shù)據(jù)，iind∈表示di存儲在ni上，表示把mi調(diào)度到ni執(zhí)行。假設(shè)mi， mj和mk的輸入數(shù)據(jù)分別存儲在nx，ny和nz上，且分別距節(jié)點(diǎn)na，nx和ny最近。當(dāng)各節(jié)點(diǎn)按照(na，nx，ny…)的順序請求任務(wù)時(shí)，基于PULL機(jī)制的方法調(diào)度任務(wù)的結(jié)果如下：，和，調(diào)度過程如圖1所示。由于mi，mj和mk的輸入數(shù)據(jù)都沒有存儲在執(zhí)行節(jié)點(diǎn)，這些任務(wù)在執(zhí)行過程中都會引起遠(yuǎn)程數(shù)據(jù)訪問延時(shí)，影響系統(tǒng)性能。

在實(shí)踐中發(fā)現(xiàn)，這種情況通常出現(xiàn)在作業(yè)執(zhí)行末尾。當(dāng)作業(yè)規(guī)模較小時(shí)，這一情況尤為嚴(yán)重。根據(jù)文獻(xiàn)[7]中的統(tǒng)計(jì)結(jié)果，在一個(gè)應(yīng)用于實(shí)際生產(chǎn)的數(shù)據(jù)中心內(nèi)部，作業(yè)平均包含的map任務(wù)數(shù)也只有42個(gè)，即大部分作業(yè)的規(guī)模都較小。如果節(jié)點(diǎn)只執(zhí)行輸入數(shù)據(jù)存儲在本地的任務(wù)，在執(zhí)行完當(dāng)前作業(yè)中此類任務(wù)的情況下，繼續(xù)執(zhí)行下個(gè)作業(yè)中的此類任務(wù)，那么即不會產(chǎn)生遠(yuǎn)程數(shù)據(jù)訪問延時(shí)，又不會浪費(fèi)計(jì)算資源。

2　基于PUSH機(jī)制的任務(wù)調(diào)度方法

為克服PULL機(jī)制存在的不足，本文提出基于PUSH機(jī)制的任務(wù)調(diào)度方法。與基于PULL機(jī)制的方法不同，該方法在調(diào)度任務(wù)時(shí)不考慮節(jié)點(diǎn)當(dāng)前是否有空閑資源，而只根據(jù)輸入數(shù)據(jù)分布推送任務(wù)。當(dāng)節(jié)點(diǎn)資源空閑時(shí)，便可開始執(zhí)行推送給自己的任務(wù)。由于只將任務(wù)推送到存儲其輸入數(shù)據(jù)的節(jié)點(diǎn)，任務(wù)在執(zhí)行過程中可以從本地磁盤訪問數(shù)據(jù)，避免了遠(yuǎn)程數(shù)據(jù)訪問延時(shí)。當(dāng)數(shù)據(jù)有多個(gè)副本且分別存儲在多個(gè)節(jié)點(diǎn)時(shí)，須同時(shí)將任務(wù)推送到這些節(jié)點(diǎn)。但是為保證效率，最終只允許任務(wù)在效率最高的節(jié)點(diǎn)執(zhí)行。

2.1任務(wù)推送

基于任務(wù)推送進(jìn)行調(diào)度是本文方法的核心，是保證把任務(wù)調(diào)度到輸入數(shù)據(jù)所在節(jié)點(diǎn)執(zhí)行的關(guān)鍵。在進(jìn)行推送時(shí)，首先根據(jù)任務(wù)的輸入數(shù)據(jù)分布，計(jì)算任務(wù)與節(jié)點(diǎn)之間的推送關(guān)系；然后，依據(jù)這一關(guān)系依次將作業(yè)中各個(gè)任務(wù)推送的相應(yīng)節(jié)點(diǎn)。本文假設(shè)每個(gè)數(shù)據(jù)都有3個(gè)副本，若存在，即的3個(gè)副本分別存儲在節(jié)點(diǎn)nt，nw和nx，則根據(jù)基于Push機(jī)制的調(diào)度方法有，和；此處， →· 表示推送，即應(yīng)推送mi到節(jié)點(diǎn)nt，nw和nx。

假設(shè)某作業(yè)包含的任務(wù)數(shù)為Sm，這些任務(wù)的輸入數(shù)據(jù)存儲在節(jié)點(diǎn)集合N中，記N={n1,n2,…,nSn}。如果以任務(wù)為單位進(jìn)行推送，則需要(3Sm)次才能將所有任務(wù)推送到輸入數(shù)據(jù)所在節(jié)點(diǎn)。Nn∈?i，假設(shè)ni存儲了個(gè)任務(wù)的輸入數(shù)據(jù)，在以任務(wù)為單位推送的前提下，需要次才能將這些任務(wù)推送到ni。此處，任務(wù)數(shù)滿足式(1)給出的約束條件。

通過分解式(1)，可知iα的取值應(yīng)在式(2)和式(3)定義的范圍之內(nèi)，同時(shí)還應(yīng)滿足式(4)定義的約束條件。

記Mi為輸入數(shù)據(jù)存儲在節(jié)點(diǎn)ni上的任務(wù)集合，且；記Mi中各任務(wù)與ni間的推送關(guān)系構(gòu)成的集合為Ri，則有。Ri中各關(guān)系式可進(jìn)行如下化簡：

故Ri亦可表示為。由Ri可知，如果以節(jié)點(diǎn)為單位進(jìn)行推送，可通過一次操作將推送到節(jié)點(diǎn)ni。輸入數(shù)據(jù)存儲在此節(jié)點(diǎn)的任務(wù)越多，該推送方式效率越高；也即越大，該推送方式效率越高。為提高效率，本文采取以節(jié)點(diǎn)為單位的推送方式，即每次只向1個(gè)節(jié)點(diǎn)推送，且1次推送當(dāng)前作業(yè)中輸入數(shù)據(jù)存儲在此節(jié)點(diǎn)上的全部任務(wù)。

推送到同一個(gè)節(jié)點(diǎn)的任務(wù)彼此競爭計(jì)算資源。為便于管理，節(jié)點(diǎn)根據(jù)系統(tǒng)采用的調(diào)度策略，建立任務(wù)隊(duì)列，根據(jù)隊(duì)列和隊(duì)列中任務(wù)的優(yōu)先級進(jìn)行本地調(diào)度。以ni為例，假設(shè)有q個(gè)優(yōu)先級，分別為0，β，…，(q-1)β，這些優(yōu)先級對應(yīng)的隊(duì)列分別記為Q(0)，。記mi2的優(yōu)先級為P(mi2)，則mi2推送到節(jié)點(diǎn)后，應(yīng)入隊(duì)列。當(dāng)ni有空閑資源時(shí)，優(yōu)先從Q(0)選擇任務(wù)執(zhí)行。只有在Q(0)為空時(shí)，才依次從其他隊(duì)列選擇任務(wù)。

2.2請求執(zhí)行

為確保在部分節(jié)點(diǎn)失效的情況下數(shù)據(jù)仍然可用，Hadoop MapReduce為每個(gè)數(shù)據(jù)都創(chuàng)建了多份副本，分別存儲在多個(gè)不同的節(jié)點(diǎn)。在這一前提下，任一任務(wù)都會被推送到多個(gè)不同的節(jié)點(diǎn)。為保證任務(wù)只在1個(gè)節(jié)點(diǎn)上執(zhí)行，特規(guī)定當(dāng)計(jì)算節(jié)點(diǎn)具備執(zhí)行某個(gè)任務(wù)的條件時(shí)，須先向管理節(jié)點(diǎn)發(fā)送執(zhí)行請求。當(dāng)多個(gè)節(jié)點(diǎn)請求執(zhí)行同一個(gè)任務(wù)時(shí)，只允許效率較高的節(jié)點(diǎn)執(zhí)行此任務(wù)。此處認(rèn)為請求較早到達(dá)的節(jié)點(diǎn)，執(zhí)行效率較高。

基于PUSH機(jī)制的方法響應(yīng)任務(wù)請求的核心算法如下：

算法1　HandleREQ(n,m,R)算法Algorithm 1　HandleREQ(n,m,R)algorithm

2.3錯(cuò)誤恢復(fù)

由于硬件、軟件等多種原因，任務(wù)在執(zhí)行過程中難免失敗。如果仍將失敗任務(wù)推送到輸入數(shù)據(jù)所在節(jié)點(diǎn)執(zhí)行，由于節(jié)點(diǎn)可能暫時(shí)沒有可用資源，失敗任務(wù)要等待較長時(shí)間才能獲得執(zhí)行機(jī)會，從而影響整個(gè)作業(yè)的執(zhí)行進(jìn)度。為避免這種情況發(fā)生，應(yīng)盡早執(zhí)行失敗任務(wù)。在任務(wù)失敗后，最先請求執(zhí)行任務(wù)的節(jié)點(diǎn)是最先有可用資源的節(jié)點(diǎn)。將失敗任務(wù)調(diào)度到此節(jié)點(diǎn)，會比調(diào)度到其他節(jié)點(diǎn)更早獲得執(zhí)行機(jī)會。

引入失敗任務(wù)處理機(jī)制后，基于PUSH機(jī)制的方法響應(yīng)任務(wù)請求的核心算法描述如下：

算法2　HandleReqWithFailure(n,m,R)算法Algorithm 2　HandleReqWithFailure(n,m,R)algorithm

2.4算法分析

文中方法利用網(wǎng)絡(luò)帶寬資源將任務(wù)推送到存儲其輸入數(shù)據(jù)副本的各個(gè)節(jié)點(diǎn)。在任務(wù)數(shù)量一定的前提下，數(shù)據(jù)副本越多，推送的任務(wù)越多，消耗的網(wǎng)絡(luò)資源也越多。記表示傳送單個(gè)任務(wù)的輸入數(shù)據(jù)所消耗的網(wǎng)絡(luò)帶寬資源，表示推送單個(gè)任務(wù)到單個(gè)節(jié)點(diǎn)消耗的網(wǎng)絡(luò)帶寬，l和lˊ分別表示采用基于PULL和基于PUSH機(jī)制調(diào)度任務(wù)時(shí)在輸入數(shù)據(jù)所在節(jié)點(diǎn)執(zhí)行的任務(wù)數(shù)，r表示數(shù)據(jù)副本個(gè)數(shù)，表示因采用文中方法產(chǎn)生的網(wǎng)絡(luò)帶寬收益，則可根據(jù)下式計(jì)算：

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

文中方法已在Hadoop-0.20.2中實(shí)現(xiàn)。為驗(yàn)證方法的有效性，將Handoop-0.20.2的原始版本和實(shí)現(xiàn)本文方法的版本部署在同一個(gè)集群上，通過對比作業(yè)在這2個(gè)Hadoop環(huán)境中的執(zhí)行情況，來評價(jià)本文方法的有效性。實(shí)驗(yàn)中用到的集群包括4個(gè)節(jié)點(diǎn)，其中1個(gè)作為管理節(jié)點(diǎn)，另外3個(gè)作為計(jì)算和存儲節(jié)點(diǎn)。表1所示為各節(jié)點(diǎn)的配置，節(jié)點(diǎn)類型1為管理節(jié)點(diǎn)的配置，節(jié)點(diǎn)類型2為計(jì)算和存儲節(jié)點(diǎn)的配置。

表1　集群配置信息Table 1　Cluster configuration

在Hadoop集群中，節(jié)點(diǎn)擁有的map/reduce slot數(shù)表示該節(jié)點(diǎn)可同時(shí)執(zhí)行的最大map/reduce任務(wù)數(shù)。由于各節(jié)點(diǎn)的硬件配置相同，故可同時(shí)執(zhí)行的最大map/reduce任務(wù)數(shù)也相同，即各個(gè)節(jié)點(diǎn)最多可同時(shí)執(zhí)行16個(gè)map任務(wù)、最多能執(zhí)行1個(gè)reduce任務(wù)。Hadoop分布式文件系統(tǒng)負(fù)責(zé)存儲作業(yè)的輸入數(shù)據(jù)。它將作業(yè)的輸入數(shù)據(jù)劃分成文件塊，分別存儲在不同的節(jié)點(diǎn)上。在本次實(shí)驗(yàn)中，設(shè)定文件塊的大小為64 MB，各個(gè)數(shù)據(jù)塊具有的副本數(shù)為3。

文中算法將map任務(wù)推送到輸入數(shù)據(jù)副本所在的各個(gè)節(jié)點(diǎn)，使其在執(zhí)行過程中可以從本地磁盤讀取數(shù)據(jù)，從而避免了遠(yuǎn)程數(shù)據(jù)訪問延時(shí)。數(shù)據(jù)副本越多，適合map任務(wù)執(zhí)行的節(jié)點(diǎn)越多，選擇高效節(jié)點(diǎn)時(shí)余地更大，但是任務(wù)推送消耗的網(wǎng)絡(luò)帶寬也越大。此外，副本越多，占用的磁盤空間越大，可用空間越少。綜合考慮，在本次實(shí)驗(yàn)中，采用了Hadoop文件系統(tǒng)推薦的設(shè)置，即為每個(gè)數(shù)據(jù)塊設(shè)置了3個(gè)副本。

在本次實(shí)驗(yàn)中跟蹤測試了6個(gè)不同的作業(yè)，作業(yè)信息如表2所示。為了更接近真實(shí)情況，選擇不同規(guī)模的作業(yè)進(jìn)行測試。在這些作業(yè)中，既有map任務(wù)數(shù)大于系統(tǒng)中map slot總數(shù)的作業(yè)，也有小于map slot總數(shù)的作業(yè)，且作業(yè)包含的map任務(wù)平均數(shù)接近文獻(xiàn)[7]在生產(chǎn)集群中的統(tǒng)計(jì)結(jié)果。在實(shí)驗(yàn)過程中，分別在2 個(gè)Hadoop環(huán)境中多次運(yùn)行這些作業(yè)，并且記錄了每次的運(yùn)行信息。通過對比各個(gè)作業(yè)在不同環(huán)境中的數(shù)據(jù)局部性、執(zhí)行時(shí)間等指標(biāo)來驗(yàn)證本文方法的有效性。

表2　測試作業(yè)信息Table 2　Details of tested jobs

本文方法擬通過改善任務(wù)的數(shù)據(jù)局部性來避免遠(yuǎn)程數(shù)據(jù)訪問延時(shí)，進(jìn)而提高Hadoop的性能。當(dāng)任務(wù)在輸入數(shù)據(jù)所在節(jié)點(diǎn)執(zhí)行時(shí)，具有最佳數(shù)據(jù)局部性，在執(zhí)行過程中不會引起遠(yuǎn)程數(shù)據(jù)訪問延時(shí)。圖2所示為各作業(yè)在不同Hadoop環(huán)境中執(zhí)行時(shí)具有最佳數(shù)據(jù)局部性的任務(wù)數(shù)。由圖2可知：在實(shí)現(xiàn)文中方法的環(huán)境(新環(huán)境)中執(zhí)行時(shí)，作業(yè)2，4，5及6包含的所有任務(wù)都具有最佳數(shù)據(jù)局部性，達(dá)到了理想狀況。

圖2　作業(yè)中具有最佳數(shù)據(jù)局部性的任務(wù)數(shù)量Fig.2　Total numbers of tasks with the best data locality

圖3　具有最佳數(shù)據(jù)局部性的任務(wù)比例圖Fig.3　Ratio of the tasks with the best data locality

圖3所示為作業(yè)中具有最佳數(shù)據(jù)局部性的任務(wù)數(shù)占總?cè)蝿?wù)數(shù)的比例。由圖3可知：當(dāng)作業(yè)在新環(huán)境中執(zhí)行時(shí)，若不存在失敗任務(wù)，則這一比例可達(dá)到100%。即使存在失敗，這一比例最低也在93%以上。而當(dāng)作業(yè)在原環(huán)境中執(zhí)行時(shí)，在最好情況下，這一比例僅達(dá)到91%，在最差情況下低至81%。由此不難看出，采用文中方法調(diào)度任務(wù)可以提高任務(wù)的數(shù)據(jù)局部性。在最好情況下，數(shù)據(jù)局部性可以提高19%左右；在最差情況下也可提高9%。

圖4所示為作業(yè)在2個(gè)Hadoop環(huán)境中執(zhí)行的平均時(shí)間。與原始方法的環(huán)境相比，當(dāng)作業(yè)1，2，4，5 和6在新環(huán)境中運(yùn)行時(shí)，執(zhí)行時(shí)間顯著降低。其中，作業(yè)5的執(zhí)行時(shí)間降低的幅度最大，達(dá)到了14.3%。作業(yè)1的執(zhí)行時(shí)間降低的幅度最小，但也超過了10%。作業(yè)3中有多個(gè)任務(wù)執(zhí)行失敗，為保證作業(yè)成功完成，系統(tǒng)不得不對這些任務(wù)進(jìn)行重新調(diào)度，消耗了過多的時(shí)間，導(dǎo)致該作業(yè)在實(shí)現(xiàn)文中方法的新環(huán)境中執(zhí)行時(shí)所用時(shí)間比原環(huán)境更長。

圖4　作業(yè)的執(zhí)行時(shí)間Fig.4　Execution time of jobs

文中引用的其他調(diào)度算法雖然為實(shí)現(xiàn)不同的調(diào)度目標(biāo)而采用了不同的調(diào)度策略，并各有長處，但在調(diào)度作業(yè)中包含的具體任務(wù)時(shí)用的都是Hadoop提供的缺省方法，即基于PULL機(jī)制的任務(wù)調(diào)度方法。故在本次實(shí)驗(yàn)中，僅對比了本文提出的基于PUSH機(jī)制的調(diào)度方法和Hadoop提供的基于PULL機(jī)制的方法。

文中方法通過將任務(wù)推送到輸入數(shù)據(jù)所在節(jié)點(diǎn)執(zhí)行來避免遠(yuǎn)程數(shù)據(jù)訪問延時(shí)。當(dāng)任務(wù)在此類節(jié)點(diǎn)執(zhí)行時(shí)，可以直接從本地磁盤訪問數(shù)據(jù)，避免了跨網(wǎng)絡(luò)的數(shù)據(jù)傳輸，節(jié)省了網(wǎng)絡(luò)資源。由于要推送任務(wù)到不同節(jié)點(diǎn)，該方法也會消耗網(wǎng)絡(luò)資源，且輸入數(shù)據(jù)副本越多，消耗的網(wǎng)絡(luò)帶寬資源也越多。在本次實(shí)驗(yàn)中，根據(jù)式(5)計(jì)算文中方法所帶來的網(wǎng)絡(luò)帶寬收益。式(5)中各參數(shù)的取值和計(jì)算結(jié)果如表3所示。

表3中wtask通過如下方式獲?。河?jì)算一組任務(wù)創(chuàng)建前后jvm中可用內(nèi)存容量之差，記該差值與該組任務(wù)總數(shù)的比值為wtask。在所有作業(yè)中，作業(yè)3由于失敗任務(wù)過多，失去了比較意義，故未計(jì)算其對應(yīng)的Wbenefit。除此之外，表3中其他作業(yè)對應(yīng)的Wbenefit遠(yuǎn)遠(yuǎn)大于1。由表3可知：文中方法通過將任務(wù)推送到輸入數(shù)據(jù)節(jié)點(diǎn)執(zhí)行，不僅可以提高系統(tǒng)性能，還可以節(jié)省網(wǎng)絡(luò)帶寬。

表3　新方法產(chǎn)生的網(wǎng)絡(luò)帶寬收益Table 3　Network width benefit from new method

4　結(jié)論

1)分析了Hadoop環(huán)境中的任務(wù)調(diào)度方法，其采用的基于PULL的調(diào)度機(jī)制無法保證把任務(wù)調(diào)度到輸入數(shù)據(jù)所在節(jié)點(diǎn)執(zhí)行，從而在作業(yè)執(zhí)行過程中引入不確定的遠(yuǎn)程數(shù)據(jù)訪問延時(shí)，影響系統(tǒng)性能。

2)提出了一種基于PUSH機(jī)制的任務(wù)調(diào)度方法，根據(jù)輸入數(shù)據(jù)分布情況，主動將任務(wù)推送到數(shù)據(jù)所在節(jié)點(diǎn)執(zhí)行，確保任務(wù)在執(zhí)行過程中可以從本地讀取數(shù)據(jù)，從而避免了遠(yuǎn)程數(shù)據(jù)訪問延時(shí)。

3)在沒有任務(wù)執(zhí)行失敗的情況下，該方法在最好情況下可將作業(yè)執(zhí)行時(shí)間降低14%左右。

[1]DEAN J,GHEMAWAT S.Mapreduce:simplified data proces -sing on large clusters[J].Communications of the ACM,2008, 51(1):107-113.

[2]The Apache Software Foundation.MapReduce tutorial[EB/OL]. [2013-08-04].https://hadoop.apache.org/docs/r1.2.1/mapred_ tutorial.html

[3]涂金金,楊明,郭麗娜.基于MapReduce的基因讀段定位算法[J].模式識別與人工智能,2014,27(3):206-212. TU Jinjin,YANG Ming,GUO Lina.Gene read mapping algorithms based on MapReduce[J].Pattern Recognition and Artificial Intelligence,2014,27(3):206-212.

[4]唐穎峰,陳世平.一種基于后綴項(xiàng)表的并行閉頻繁項(xiàng)集挖掘算法[J].計(jì)算機(jī)應(yīng)用研究,2014,31(2):373-377. TANG Yingfeng,CHEN Shiping.Parallel closed frequent itemset mining algorithm with post fix-table[J].Application Research of Computers,2014,31(2):373-377.

[5]王曉佳,楊善林,陳志強(qiáng).大數(shù)據(jù)時(shí)代下的情報(bào)分析與挖掘技術(shù)研究:電信客戶流失情況分析[J].情報(bào)學(xué)報(bào),2013,32(6): 564-574. WANG Xiaojia,YANG Shanlin,CHEN Zhiqiang.Research on information analysis and data mining in the age of big data: analysis of customer loss in telecom[J].Journal of the China Society for Scientific and Technical Information,2013,32(6): 564-574.

[6]付天新,劉正軍,閆浩文.基于MapReduce模型的生物量遙感并行反演方法研究[J].干旱區(qū)資源與環(huán)境,2013,27(1): 130-136. FU Tianxin,LIU Zhengjun,YAN Haowen.Remote sensing retrieval method for biomass based on MapReduce parallel model[J].Journal of Arid Land Resource and Environment,2013, 27(1):130-136.

[7]REN Zujie,WAN Jian,SHI Weisong.Workload analysis, implications and optimization on a production Hadoop cluster:a casestudyontaobao[J].IEEETransactionsonServices Computing,2014,7(2):307-321.

[8]ZAHARIA M,BORTHAKUR D,SARMA S J,et al.Delay scheduling:a simple technique for achieving locality and fairness in cluster scheduling[C]//Proc of the 5th European Conference on Computer Systems.New York:ACM,2010: 265-273.

[9]ZHANG Xiaohong,ZHONG Zhiyong,FENG Shengzhong,et al. Improvingdatalocalityofmapreducebyschedulingin homogeneouscomputingenvironments[C]//IEEE9th International Symposium on Parallel and Distributed Processing with Applications,Washington DC:IEEE,2011:120-126.

[10]WANG Weina,ZHU Kai,YING Lei,et al.A throughput optimal algorithm for map task scheduling in MapReduce with data locality[J].ACM SIGMETRICS Performance Evaluation Review, 2013,40(4):33-42.

[11]TANG Zhuo,ZHOU Junqing,LI Kenli,et al.A map-reduce task schedulingalgorithmfordeadlineconstraints[J].Cluster Computing,2013,16(4):651-662.

[12]TAN Jian,MENG Xiaoqiao,ZHANG Li.Coupling task progress for MapReduce resource-aware scheduling[C]//Proc of IEEE INFOCOM,Washington DC:IEEE,2013:1618-1626.

[13]LU Peng,LEE Youngchoon,WANG Chen,et al.Workload characteristic oriented scheduler for MapReduce[C]//Proc of the 2012 IEEE 18th International Conference on Parallel and Distributed Systems,Washington DC:IEEE,2012:156-163.

[14]MASHAYEKHYL,NEJADMN,GROSUD,etal. Energy-aware scheduling of MapReduce jobs for big data application[J].IEEE Transaction on Parallel and Distributed Systems,2015,26(10):2720-2733.

[15]SUN Mingming,ZHUANG Hang,ZHOU Xuehai,et al.HPSO: perfecting basedschedulingtoimprovedatalocalityfor MapReduce clusters[C]//Proc of 14th International Conference on Algorithms and Architectures for Parallel Processing,Cham: Springer International Publishing,2014:82-95.

[16]KAVULYA S,TAN J,GANDHI R.An analysis of traces from a productionMapReducecluster[C]//ProcofIEEE/ACM International Conference on Cluster,Cloud and Grid Computing. Washington DC:IEEE,2010:94-103.

(編輯羅金花)

Ascheduling method based on task pushing in MapReduce

ZHANG Xiaohong1,2,3,SUN Jiangfeng1,3,ZHAO Wentao1,3
(1.School of Computer Science and Technology,Henan Polytechnic University,Jiaozuo 454003,China;
2.Shenzhen Institutes ofAdvanced Technology,Chinese Academy of Sciences,Shenzhen 518055,China;
3.Provical Open Laboratory of Mine Informatization Key Discipline,Jiaozuo 454003,China)

To reduce remote data access latency and improve the system performance in Hadoop MapReduce,a new task scheduling method was proposed.According to the method,tasks were pushed to the nodes of storing their input data. When executing on those nodes,those tasks can access the relative input data from local disks,and hence avoiding remote data access latency.The new method was implemented in Hadoop-0.20.2,and evaluated in a real cluster.The results show that the method can decrease the execution time of jobs by 14.3%in the best case,and 8%on average.

data locality;performance optimization;task scheduling;MapReduce

趙文濤，教授，碩士生導(dǎo)師，從事分布式計(jì)算技術(shù)、大數(shù)據(jù)技術(shù)研究；E-mail:zwt@hpu.edu.cn

TP315

A

1672-7207(2016)07-2334-07

10.11817/j.issn.1672-7207.2016.07.022

2015-07-23；

2015-09-23

國家自然科學(xué)基金面上資助項(xiàng)目(51274088)；河南省教育廳項(xiàng)目(ITE12103)；河南理工大學(xué)礦山信息化省級重點(diǎn)實(shí)驗(yàn)室項(xiàng)目(KY2012-05)；河南理工大學(xué)博士基金資助項(xiàng)目(B2012-099)；河南省科技攻關(guān)項(xiàng)目(142102210435)(Project(51274088)supported by the National Natural Science Foundation of China;Project(ITE12103)supported by the Foundation of Henan Educational Committee; Project(KY2012-05)supported by the Foundation of Provincial Open Laboratory of Mine Informatization Key Discipline;Project(B2012-099) supported by the PhD Foundation of Henan Polytechnic University;Project(142102210435)supported by the Programs for Science and Technology Development of Henan Province)