王力生　魏薇

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

關鍵詞：MapReduce；調度算法；優(yōu)化；容錯性；推測性執(zhí)行

中圖分類號：TP316.4 文獻標識碼：A 文章編號：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 概述

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

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

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

2 MapReduce框架介紹

2.1 MapReduce工作原理

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

以MapReduce的開源實現(xiàn)Hadoop[3]為例，計算任務的執(zhí)行過程如圖1所示。

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

2） 執(zhí)行Map任務的節(jié)點以數(shù)據(jù)塊作為輸入，調用用戶定義的Map函數(shù)，把輸出的中間結果寫入內存緩沖區(qū)。當內存緩沖區(qū)快要寫滿時，再把數(shù)據(jù)寫到本地磁盤。

3） 在寫入磁盤之前，Map節(jié)點會對數(shù)據(jù)進行排序和歸并，按照鍵值把數(shù)據(jù)映射到不同的分區(qū)，使每個分區(qū)的數(shù)據(jù)對應一個Reduce任務，再通過JobTracker節(jié)點通知Reduce節(jié)點數(shù)據(jù)的存儲位置。

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

2.2 MapReduce在異構環(huán)境中存在的問題

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

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

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

3.1 LATE算法

文獻[5]提出了一種名為LATE算法的任務調度策略。LATE算法的核心思想是對執(zhí)行結束時間最長的任務進行備份，因為這樣的任務最有可能拖慢整個計算任務的響應時間。假設Timei為Ti的已執(zhí)行時間，PSi為Ti的任務進度值，對任務Ti的結束時間TRi的估計，LATE算法采用以下公式：

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

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

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

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

3.2 SAMR算法

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

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

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

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

3.3 ESAMR算法

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

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

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

4 總結

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

綜上所述，對于MapReduce在異構集群環(huán)境下的調度算法仍然有待優(yōu)化，是一個充滿前途的挑戰(zhà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.