王力生　魏薇

摘要：MapReduce作為一個(gè)分布式并行計(jì)算框架，在大數(shù)據(jù)處理方面得到了廣泛的應(yīng)用。該計(jì)算框架在同構(gòu)集群環(huán)境中能夠高效地運(yùn)行，但是在異構(gòu)集群環(huán)境中原容錯(cuò)算法不能正確地檢測(cè)慢速任務(wù)，導(dǎo)致了性能的大幅下降。該文針對(duì)這一現(xiàn)象，分析了問(wèn)題的主要原因，并且介紹了現(xiàn)存的幾個(gè)優(yōu)化算法，即Longest Approximate Time to End（LATE）算法，Self-Adaptive MapReduce（SAMR）算法，Enhanced Self-Adaptive MapReduce（ESAMR）算法，比較了各個(gè)算法的優(yōu)缺點(diǎn)，最后指出了未來(lái)的研究方向。

關(guān)鍵詞：MapReduce；調(diào)度算法；優(yōu)化；容錯(cuò)性；推測(cè)性執(zhí)行

中圖分類(lèi)號(hào)：TP316.4 文獻(xiàn)標(biāo)識(shí)碼：A 文章編號(hào)：1009-3044（2015）01-0051-03

Survey on MapReduce Scheduling Algorithms in Heterogeneous Environments

WANG Li-Sheng，WEI Wei

（Department of Electronic and Information Engineering， Tongji University， Shanghai 201804， China）

Abstract： As a parallel programming model， MapReduce is widely used to process large data sets on a cluster. The current MapReduce implementation works effectively in homogeneous environment， but has a poor performance due to the static method used to detect stragglers. This paper discusses how the heterogeneity affects the MapReduce performance and surveys some of the approaches that have been designed to improve the MapReduce performance in heterogeneous environments. Advantages and disadvantages of these algorithms are identified.

Key words： MapReduce； scheduling algorithms； optimization； fault tolerance； speculative execution

1 概述

近年來(lái)，隨著互聯(lián)網(wǎng)技術(shù)的迅猛發(fā)展，越來(lái)越多的網(wǎng)絡(luò)應(yīng)用需要進(jìn)行大數(shù)據(jù)的處理和存儲(chǔ)。為了滿足計(jì)算需求，計(jì)算資源逐漸由單機(jī)多核發(fā)展為集群眾核。MapReduce[1， 2]是由Google提出的一個(gè)用于海量數(shù)據(jù)處理的分布式并行計(jì)算框架，在大數(shù)據(jù)處理方面得到了業(yè)內(nèi)的廣泛認(rèn)可。大多數(shù)互聯(lián)網(wǎng)公司都使用MapReduce來(lái)處理大數(shù)據(jù)的查詢(xún)響應(yīng)以及數(shù)據(jù)挖掘工作。

MapReduce框架最初被設(shè)計(jì)在同構(gòu)環(huán)境中運(yùn)行，即各檢點(diǎn)的計(jì)算性能、存儲(chǔ)容量、存儲(chǔ)速度和網(wǎng)絡(luò)帶寬是相近的。MapReduce在進(jìn)行輸入數(shù)據(jù)劃分、集群任務(wù)調(diào)度和容錯(cuò)性處理時(shí)，也都是基于同構(gòu)環(huán)境的性質(zhì)做出決策。但是隨著集群規(guī)模的擴(kuò)展，保持所有節(jié)點(diǎn)都屬于同一機(jī)型是相當(dāng)困難的事，所以MapReduce框架也可能會(huì)被部署在異構(gòu)環(huán)境中，即各節(jié)點(diǎn)的計(jì)算性能、存儲(chǔ)速度等方面存在較大的差異。

由于最初設(shè)計(jì)時(shí)沒(méi)有充分考慮異構(gòu)環(huán)境的運(yùn)行情況，MapReduce在異構(gòu)環(huán)境中的性能并不理想。針對(duì)這一問(wèn)題，國(guó)內(nèi)外的一些學(xué)者分析了MapReduce性能下降的原因，并且提出了一些異構(gòu)環(huán)境下的改進(jìn)算法。該文通過(guò)對(duì)這些算法進(jìn)行分析，總結(jié)出各個(gè)算法的優(yōu)缺點(diǎn)，希望以此作為相關(guān)技術(shù)人員的參考。

2 MapReduce框架介紹

2.1 MapReduce工作原理

將海量數(shù)據(jù)分成較小的數(shù)據(jù)塊，分發(fā)到各個(gè)節(jié)點(diǎn)并行處理。對(duì)用戶(hù)而言，任務(wù)在分布式集群中的調(diào)度過(guò)程是透明的。用戶(hù)只需要實(shí)現(xiàn)Map函數(shù)和Reduce函數(shù)即可，其中，Map函數(shù)處理輸入的鍵值對(duì)（key/value），并生成一組臨時(shí)的鍵值對(duì)，發(fā)送給Reduce函數(shù)進(jìn)行處理；Reduce函數(shù)處理臨時(shí)鍵值對(duì)，生成最終結(jié)果寫(xiě)到分布式文件系統(tǒng)。Map函數(shù)和Reduce函數(shù)的并行調(diào)度由MapReduce框架完成。

以MapReduce的開(kāi)源實(shí)現(xiàn)Hadoop[3]為例，計(jì)算任務(wù)的執(zhí)行過(guò)程如圖1所示。

1） 主節(jié)點(diǎn)JobTracker將存儲(chǔ)在Hadoop分布式文件系統(tǒng)[4]（HDFS）上的用戶(hù)輸入數(shù)據(jù)劃分為若干份，每一份數(shù)據(jù)的大小約64M（可通過(guò)配置文件設(shè)置），并為每一個(gè)劃分創(chuàng)建一個(gè)map任務(wù)，分配給從節(jié)點(diǎn)TaskTracker執(zhí)行。

2） 執(zhí)行Map任務(wù)的節(jié)點(diǎn)以數(shù)據(jù)塊作為輸入，調(diào)用用戶(hù)定義的Map函數(shù)，把輸出的中間結(jié)果寫(xiě)入內(nèi)存緩沖區(qū)。當(dāng)內(nèi)存緩沖區(qū)快要寫(xiě)滿時(shí)，再把數(shù)據(jù)寫(xiě)到本地磁盤(pán)。

3） 在寫(xiě)入磁盤(pán)之前，Map節(jié)點(diǎn)會(huì)對(duì)數(shù)據(jù)進(jìn)行排序和歸并，按照鍵值把數(shù)據(jù)映射到不同的分區(qū)，使每個(gè)分區(qū)的數(shù)據(jù)對(duì)應(yīng)一個(gè)Reduce任務(wù)，再通過(guò)JobTracker節(jié)點(diǎn)通知Reduce節(jié)點(diǎn)數(shù)據(jù)的存儲(chǔ)位置。

4） 執(zhí)行Reduce任務(wù)的節(jié)點(diǎn)從遠(yuǎn)程獲取到數(shù)據(jù)以后，對(duì)來(lái)自不同Map節(jié)點(diǎn)的數(shù)據(jù)進(jìn)行排序和歸并，然后調(diào)用用戶(hù)定義的Reduce函數(shù)，得到最終結(jié)果后寫(xiě)入文件系統(tǒng)。

2.2 MapReduce在異構(gòu)環(huán)境中存在的問(wèn)題

MapReduce在分配計(jì)算任務(wù)時(shí)基于集群同構(gòu)的前提進(jìn)行決策，分配給每臺(tái)機(jī)器的任務(wù)槽數(shù)量和計(jì)算數(shù)據(jù)是基本相同的。同時(shí)，出于容錯(cuò)方面的考慮，為了防止執(zhí)行速度慢的任務(wù)影響整體的執(zhí)行進(jìn)度，MapReduce使用推測(cè)執(zhí)行機(jī)制，會(huì)為慢速任務(wù)啟動(dòng)一個(gè)備份任務(wù)，讓備份任務(wù)與原始任務(wù)同時(shí)處理同一份數(shù)據(jù)，將先運(yùn)行完的任務(wù)的輸出作為最終結(jié)果。其中，慢速任務(wù)通過(guò)任務(wù)進(jìn)度值（ProgressScore）來(lái)評(píng)估，范圍是0到1之間的小數(shù)。對(duì)于Map任務(wù)，任務(wù)進(jìn)度值為輸入數(shù)據(jù)的處理比例；對(duì)于Reduce任務(wù)，任務(wù)被分為復(fù)制數(shù)據(jù)、排序和執(zhí)行用戶(hù)定義的Reduce函數(shù)三個(gè)階段，每個(gè)階段各占1/3。假設(shè)M表示任務(wù)Ti已處理的鍵值對(duì)數(shù)，N表示任務(wù)Ti需要處理的總鍵值對(duì)數(shù)，K表示Reduce任務(wù)當(dāng)前所處的階段，則任務(wù)Ti的進(jìn)度值PSi計(jì)算公式如下：

然而，在異構(gòu)環(huán)境中，以上描述的機(jī)制不能很好的執(zhí)行。因?yàn)?，高性能的?jié)點(diǎn)能夠更加快速地運(yùn)行任務(wù)，拉高了平均進(jìn)度值，從而使較多性能低的節(jié)點(diǎn)被判為慢速任務(wù)，導(dǎo)致大量任務(wù)需要備份。節(jié)點(diǎn)之間數(shù)據(jù)的傳輸增大了網(wǎng)絡(luò)通信開(kāi)銷(xiāo)，使異構(gòu)環(huán)境中的框架性能降低。

3 MapReduce調(diào)度優(yōu)化算法分析

3.1 LATE算法

文獻(xiàn)[5]提出了一種名為L(zhǎng)ATE算法的任務(wù)調(diào)度策略。LATE算法的核心思想是對(duì)執(zhí)行結(jié)束時(shí)間最長(zhǎng)的任務(wù)進(jìn)行備份，因?yàn)檫@樣的任務(wù)最有可能拖慢整個(gè)計(jì)算任務(wù)的響應(yīng)時(shí)間。假設(shè)Timei為T(mén)i的已執(zhí)行時(shí)間，PSi為T(mén)i的任務(wù)進(jìn)度值，對(duì)任務(wù)Ti的結(jié)束時(shí)間TRi的估計(jì)，LATE算法采用以下公式：

針對(duì)異構(gòu)環(huán)境中可能會(huì)出現(xiàn)大量備份任務(wù)這一問(wèn)題，LATE算法定義閾值SpeculativeCap（大約總?cè)蝿?wù)槽數(shù)的10%），表示系統(tǒng)中同時(shí)可以運(yùn)行的最大備份任務(wù)數(shù)，當(dāng)備份任務(wù)數(shù)達(dá)到閾值時(shí)，不會(huì)啟動(dòng)新的備份任務(wù)。另外，LATE算法認(rèn)為備份任務(wù)不應(yīng)該被運(yùn)行在慢節(jié)點(diǎn)上，因此定義閾值SlowNodeThreshold（大約25%），如果任務(wù)進(jìn)度值低于該閾值，則認(rèn)為當(dāng)前節(jié)點(diǎn)也是一個(gè)慢節(jié)點(diǎn)，備份任務(wù)不能在該節(jié)點(diǎn)上運(yùn)行。

LATE算法的調(diào)度策略可以總結(jié)為：當(dāng)一個(gè)節(jié)點(diǎn)出現(xiàn)空閑資源，且系統(tǒng)中總的備份任務(wù)數(shù)小于SpeculativeCap時(shí)，（1） 如果該節(jié)點(diǎn)是慢節(jié)點(diǎn)（節(jié)點(diǎn)得分低于SlowNodeThreshold），則忽略這個(gè)請(qǐng)求。（2） 對(duì)當(dāng)前正在運(yùn)行的任務(wù)按估算的剩余完成時(shí)間排序。（3） 選擇剩余完成時(shí)間最大且進(jìn)度值低于SlowTaskThreshold的任務(wù)，為該任務(wù)啟動(dòng)備份任務(wù)。

對(duì)比Hadoop內(nèi)置的調(diào)度算法，LATE算法在異構(gòu)環(huán)境中僅備份最慢的任務(wù)，并且控制了系統(tǒng)中備份任務(wù)的總數(shù)，提升了異構(gòu)環(huán)境中集群的總體性能。通過(guò)確保將執(zhí)行時(shí)間最長(zhǎng)的任務(wù)備份在快節(jié)點(diǎn)上，能夠有效地縮短任務(wù)的響應(yīng)時(shí)間。

但是，LATE算法也存在一些不足。任務(wù)結(jié)束時(shí)間的估計(jì)是建立在任務(wù)線性運(yùn)行的假設(shè)上的，通常不能正確判斷發(fā)生異常故障的節(jié)點(diǎn)。判斷也只能在任務(wù)執(zhí)行一分鐘之后進(jìn)行，不夠及時(shí)。

3.2 SAMR算法

針對(duì)LATE算法不能適應(yīng)異構(gòu)環(huán)境動(dòng)態(tài)變化的問(wèn)題，文獻(xiàn)[6]提出了名為SAMR的算法。該算法基于LATE算法的核心思想，改進(jìn)了最慢執(zhí)行任務(wù)的推測(cè)。相對(duì)于Hadoop內(nèi)置的調(diào)度算法，SAMR算法將執(zhí)行時(shí)間縮短了近24%；相對(duì)于LATE算法，執(zhí)行時(shí)間縮短了近14%。

SAMR算法認(rèn)為計(jì)算任務(wù)進(jìn)度值時(shí)，Reduce任務(wù)三個(gè)階段的執(zhí)行時(shí)間比例不能絕對(duì)地設(shè)置為1/3（即R1=R2=R3=1/3） ，Map任務(wù)的排序階段也不能直接忽視（即M1=1，M2=0），都應(yīng)該根據(jù)節(jié)點(diǎn)的性能設(shè)置為不同的值。對(duì)慢節(jié)點(diǎn)的備份也應(yīng)該分為Map慢節(jié)點(diǎn)和Reduce慢節(jié)點(diǎn)分別進(jìn)行，因?yàn)橛行┣闆r下Reduce慢節(jié)點(diǎn)沒(méi)有必要進(jìn)行備份。該算法在每個(gè)節(jié)點(diǎn)上記錄本地運(yùn)行任務(wù)的時(shí)間信息，執(zhí)行任務(wù)時(shí)讀取本地的歷史信息，動(dòng)態(tài)地調(diào)整任務(wù)進(jìn)度值中的參數(shù)值M1-2、R1-3。通過(guò)公式（4） 計(jì)算出任務(wù)Ti的ProgressRatei后，如果ProgressRatei滿足以下公式，則被判斷為慢節(jié)點(diǎn)：

其中，SlowTaskCap是0到1直接的值，越接近0，就有越多的任務(wù)被判斷為慢任務(wù)。在執(zhí)行完計(jì)算任務(wù)后，SAMR算法更新本地的歷史數(shù)據(jù)，將本次參數(shù)信息寫(xiě)入文件。

SAMR算法與LATE算法相比，能夠根據(jù)不同節(jié)點(diǎn)的性能，更準(zhǔn)確地計(jì)算任務(wù)進(jìn)度值，推測(cè)出需要備份的慢節(jié)點(diǎn)。但是，SAMR算法忽略了數(shù)據(jù)集的大小和不同計(jì)算任務(wù)類(lèi)型對(duì)任務(wù)進(jìn)度值計(jì)算參數(shù)的影響，僅依靠歷史信息調(diào)整計(jì)算參數(shù)仍然存在偏差。

3.3 ESAMR算法

ESAMR算法[7]是SAMR算法的一個(gè)優(yōu)化版本，基于記錄歷史執(zhí)行信息的方案，采用k-means算法動(dòng)態(tài)調(diào)整計(jì)算任務(wù)進(jìn)度值公式的參數(shù)M1-2、R1-3，提高了推測(cè)慢速任務(wù)的準(zhǔn)確率。

ESAMR算法根據(jù)參數(shù)M1-2、R1-3的數(shù)值，通過(guò)機(jī)器學(xué)習(xí)的技術(shù)將每個(gè)節(jié)點(diǎn)上的記錄的歷史信息劃分為k個(gè)聚簇。對(duì)于Map階段，該算法根據(jù)計(jì)算任務(wù)在節(jié)點(diǎn)上已完成的Map任務(wù)得出一個(gè)M1的臨時(shí)值，通過(guò)臨時(shí)值找到最鄰近的聚簇，使用該聚簇的任務(wù)進(jìn)度值計(jì)算節(jié)點(diǎn)的任務(wù)結(jié)束時(shí)間；對(duì)于Reduce階段，采用類(lèi)似的方法，通過(guò)R1和R2的臨時(shí)值找到最鄰近的聚簇，使用該聚簇的任務(wù)進(jìn)度值來(lái)推測(cè)慢速任務(wù)。在計(jì)算任務(wù)結(jié)束后，ESAMR算法記錄各節(jié)點(diǎn)的執(zhí)行信息，然后對(duì)聚簇進(jìn)行重新劃分。

對(duì)比于LATE算法和SAMR算法，ESAMR算法能夠跟準(zhǔn)確地推測(cè)慢速任務(wù)，從而提高了集群的運(yùn)行效率。不足之處在于，使用k-means算法本身也存在一些額外的開(kāi)銷(xiāo)，增加了系統(tǒng)的負(fù)載。

4 總結(jié)

本文以Hadoop內(nèi)置調(diào)度算法為例，介紹了MapReduce框架在異構(gòu)集群環(huán)境中性能下降的主要原因，并且對(duì)LATE算法、SAMR算法和ESAMR算法在異構(gòu)集群中的表現(xiàn)進(jìn)行了分析和比較。MapReduce框架在設(shè)計(jì)時(shí)僅考慮了同構(gòu)集群的環(huán)境，其容錯(cuò)算法在異構(gòu)集群中會(huì)導(dǎo)致大量任務(wù)備份，影響框架的整體性能。LATE算法提出備份運(yùn)行結(jié)束時(shí)間最長(zhǎng)的任務(wù)的核心思想，提高了框架的響應(yīng)時(shí)間。SAMR算法基于LATE算法的思想，利用節(jié)點(diǎn)的歷史信息更加準(zhǔn)確地計(jì)算任務(wù)的結(jié)束時(shí)間。ESAMR算法在SAMR算法的基礎(chǔ)上，通過(guò)數(shù)據(jù)挖掘的技術(shù)動(dòng)態(tài)調(diào)整計(jì)算參數(shù)，提高了結(jié)束時(shí)間計(jì)算的準(zhǔn)確性。

綜上所述，對(duì)于MapReduce在異構(gòu)集群環(huán)境下的調(diào)度算法仍然有待優(yōu)化，是一個(gè)充滿前途的挑戰(zhàn)的領(lǐng)域。

參考文獻(xiàn)：

[1] Dean J，Ghemawat S.Mapreduce： simplied data processing on large clusters[C]//OSDI 2004： Proceedings of 6th Symposium on Operating System Design and Implemention，（New York）， ACM Press，2004：137-150.

[2] Dean J，Ghemawat S.MapReduce： a flexible data processing tool[C]//Communications of the ACM，2010，53（1）：72-77.

[3] Apache Hadoop[EB/OL].http：//hadoop.apache.org.

[4] Hadoop Distributed File System[EB/OL].http：//hadoop.apache.org/hdfs.

[5] Zaharia M，Konwinski A，Joseph A D，et al.Improving mapreduce performance in heterogeneous environments[C]//8th Usenix Symposium on Operating Systems Design and Implementation， （New York）， ACM Press，2008：29-42.

[6] Quan Chen， Daqiang Zhang， Minyi Guo， et al.SAMR： A Self-adaptive MapReduce Scheduling Algorithm in Heterogeneous Environment[C].Computer and Information Technology （CIT）， 2010 IEEE 10th International Conference on， 2010：2736-2743.

[7] Xiaoyu Sun， Chen He，Ying Lu.ESAMR： An Enhanced Self-Adaptive MapReduce Scheduling Algorithm[C]//IEEE 18th International Conference on Parallel and Distributed Systems，2012.