MapReduce中數據傾斜解決方法的研究

王 剛，李盛恩

(山東建筑大學計算機科學與技術學院，山東濟南 250101)

MapReduce中數據傾斜解決方法的研究

王 剛，李盛恩

(山東建筑大學計算機科學與技術學院，山東濟南 250101)

隨著移動互聯網和物聯網的飛速發展，數據規模呈爆炸性增長態勢，人們已經進入大數據時代。MapReduce是一種分布式計算框架，具備海量數據處理的能力，已成為大數據領域研究的熱點。但是MapReduce的性能嚴重依賴于數據的分布，當數據存在傾斜時，MapReduce默認的Hash劃分無法保證Reduce階段節點負載平衡，負載重的節點會影響作業的最終完成時間。為解決這一問題，利用了抽樣的方法。在用戶作業執行前運行一個MapReduce作業進行并行抽樣，抽樣獲得key的頻次分布后結合數據本地性實現負載均衡的數據分配策略。搭建了實驗平臺，在實驗平臺上測試WordCount實例。實驗結果表明，采用抽樣方法實現的數據劃分策略性能要優于MapReduce默認的哈希劃分方法，結合了數據本地性的抽樣劃分方法的效果要優于沒有考慮數據本地性的抽樣劃分方法。

大數據;MapReduce;負載均衡;抽樣

0 引言

伴隨著互聯網、物聯網和移動互聯網的快速發展，每天會產生海量數據，數據處于爆炸式的增長狀態，這預示著大數據時代的到來。MapReduce［1］是Google提出的一種分布式計算框架。由于其具有高可擴展性、高效性和容錯性等特點，在大規模數據處理中得到了廣泛應用。用戶使用MapReduce處理海量數據時，只需根據業務邏輯編寫Map和Reduce函數即可，并行化、容錯、數據分發和負載平衡等復雜的技術由MapReduce運行時庫自動完成。Hadoop是MapReduce的開源實現，在云計算和大數據處理等領域應用廣泛，成為了研究的熱點。

性能優化的重點就是負載均衡。在MapReduce分布式計算框架下，用戶提交的作業被劃分成若干個塊，每個塊被分配給一個Mapper執行，Map階段產生的中間結果經過劃分函數交給Reducer執行并產生最終的結果。整個作業的完成時間由Reducer運行最慢的決定。當節點的負載出現不均衡時，負載重的節點會制約作業的完成時間。因此，每個節點能否在一致的時間內完成是影響分布式計算性能的關鍵因素。

文中首先通過抽樣獲取key的頻率分布信息，然后根據數據本地性特征，利用貪心策略實現Reduce節點的負載均衡。

1 相關工作

1.1 MapReduce

Google過去在處理海量數據時，采用了高配置服務集群的方法，但是隨著數據規模越來越大，傳統方法在性能方面表現不足。Google設計了新的分布式計算模型MapReduce，MapReduce可以部署在廉價的商用機器上，提高了處理海量數據的性能，降低了硬件成本。MapReduce最大的優勢就是簡單易用。

Hadoop開源實現了Google的MapReduce模型，并且提供了分布式文件系統HDFS。用戶提交作業后，數據被等大小切分給Map節點處理，Map節點執行Map函數產生中間結果并根據劃分函數保存在本地磁盤。Reduce節點讀取中間結果執行Reduce函數產生最終的結果。在這個過程中，用戶不必關注分布式處理的細節，作業調度、數據劃分以及容錯處理這些細節由MapReduce自動完成。除了Google的內部實現外，MapReduce還有一個應用廣泛的開源實現Hadoop。

1.2 已有工作

MapReduce默認的劃分方法是把哈希值相同的key分配給同一個Reducer節點，在數據傾斜的情況下，容易造成Reducer端負載不均，影響任務的完成時間。目前研究人員針對Reducer端負載不平衡做了大量的研究工作。文獻［2］提出的SkewTune系統對Hadoop進行功能增強，當有空閑節點時，系統會將當前負載最重的任務分配給空閑節點，從而縮短整個作業的執行時間。文獻［3］提出了LEEN算法。該算法設計了最優的劃分函數，通過把一個key分配到最合適的分組來實現負載均衡。這些方法都是在MapReduce運行過程中進行調整，操作復雜，通用性低。

除了修改MapReduce框架來消除負載不均外，目前還有一種常用的方法就是抽樣。文獻［4］通過抽樣把key劃分成大、中、小三種負載，劃分函數根據負載大小的不同會有不同的處理方式。文獻［5］先執行一個MapReduce作業，抽樣統計key的分布情況，從而給出自定義的劃分函數。文獻［6］基于range partition提出了改進的方法。文獻［7］在簡單采樣的基礎上提出性能更優的動態劃分方法。

基于以上研究工作，文中嘗試利用數據本地性和抽樣來完善Reduce負載均衡機制。首先通過抽樣獲取key的分布，其次理論分析Hash算法的不足，結合數據本地性提出貪心策略的Reduce端負載均衡，最后通過大規模的數據驗證算法的有效性。

2 基于抽樣的數據劃分

2.1 數據傾斜

MapReduce的性能很大程度上依賴于數據的分布［8］，如果數據分布不均勻性能就無法保證。但是科學數據往往都是存在傾斜的，MapReduce在處理傾斜數據時，Map階段的中間結果利用哈希函數分配給Reduce階段的節點，MapReduce哈希劃分:partition-Num=key.hashCode()%REDUCER_NUM。這種方法可以保證每個Reducer處理的劃分中包含的分組數目相同，但無法保證分組內部記錄總數相同，特別是在數據傾斜的情況下［9］。使用 Hash算法時，多個 key的hashcode與Reduce節點數量求余數之后可能具有相同的值，從而使數據劃分集中于某一個Reduce節點，造成數據分布不均衡。Hash算法沒有考慮key的頻次，可能存在一些頻次大的key被劃分到同一Reduce節點，造成數據不均衡［10］。

如圖1所示，有3個數據節點，Map端輸入數據有9個key值，每個key值的數據量不相等，但是每個數據節點的總量是相等的。圖中Node1的key值K3的數據量為12，表示為K3:12。計算可得Node1的數據總量為70。則由Hadoop默認的哈希劃分函數分區之后，Reduce端輸入的數據量不相等，出現了數據傾斜，三個Reducer的數據量分別為34，56，120，Reducer3的數據量比其他兩個節點的數據量多，數據傾斜會影響任務的最終完成時間。

圖1 哈希劃分不平衡示例

2.2 抽 樣

從總體單位中抽取部分單位作為樣本的方法就是抽樣［11］。其基本要求是要保證所抽取的樣品單位對全部樣品具有充分的代表性。文中算法針對大規模的數據進行處理，如果對所有的數據進行統計，成本太高，因此采用抽樣方法，獲取key的頻率分布［12］。

在抽樣前，需要總體單位有序，根據樣本容量確定抽樣間隔。假設總體單位容量為M，樣本容量為N，則抽樣間隔為K=M/N。從總體中隨機確定一個單位作為第一個樣本，然后每隔K個距離確定一個樣本單位，達到樣本容量即停止。抽取的樣本容量越多，抽樣的準確度越高。

2.3 負載均衡方法

文中提出基于采樣的方法，對Mapper的輸出結果進行采樣，增加一個mapreduce job來獲得key的分布信息。這個過程主要包括兩個步驟:

步驟1:運行一個mapreduce job獲得中間數據集樣本，然后統計key的分布，根據樣本分布產生劃分。劃分結果用一個映射數據結構表示:(k，p)鍵為k被劃分到了p。

步驟2:運行真正的數據處理任務。劃分函數根據步驟1獲得的(k，p)產生劃分策略而不再利用散列劃分。

然而，在MapReduce現有的調度策略中并未充分考慮數據本地性［13］，在任務的調度過程中只是簡單地從隊列中取出第一個待分配的任務給當前可用節點而忽略了中間數據的分布特點，因此可能導致大量的中間文件必須跨網絡傳輸到該節點，如圖2所示。

在圖2中，Partion2和Partion3都要跨網絡傳輸，增加了時間開銷。為了減少網絡傳輸帶來的開銷，減少作業運行時間，文中提出了數據本地性感知的抽樣劃分算法。

定義1:設M表示輸入數據的總量，可以用輸入數據的行數近似表示，N表示參與計算的節點數，則劃分算法應該使節點的負載接近M/N。設P表示鍵值key被劃分的分區，V表示節點已分配的數據總量，TK表示鍵為key的總記錄數，元組(key，sum，node)表示節點node上鍵key的數量為sum。

結合抽樣技術和數據本地性算法的具體執行過程如下:

Step1:在每個節點上進行抽樣，抽樣的結果形式為(key，sum，node)。

Step2:統計所有節點上鍵為key的總數，用 TK表示。

Step3:將中間結果按數量從大到小排序。

Step4:遍歷(key，sum，node)，如果(sum+V)小于M/N，則將key劃分到節點node。

Step5:遍歷處理步驟4中沒有涉及的 key，把鍵key分配到V最小的節點中。

Step6:在抽樣過程中，沒有抽取的key認為是小概率數據，不影響Reduce端的負載均衡。沒有抽取的key使用Hadoop默認的哈希劃分。

圖1的例子利用文中的負載均衡算法進行數據劃分之后的結果如圖3所示。可以看出，文中所提的利用數據本地性的抽樣方法獲得了較好的Reduce端負載均衡，優化了默認的哈希劃分和簡單的抽樣劃分。

算法1:基于數據本地性的抽樣劃分。

Input:pairs of(key，sum，node)，M，N

Output:partition result P

1.T←total rows of each key value in all node，put all key into K ，initialize the map P and list V

2.while K is not null

3.if P［key］is not partition and V［node］+TK［key］＜=M/N

4.P.add(key，node)

5.V［node］+=TK［key］

6.remove the key from K

7.end if

8.end while

9.while K is not null

10.node←search minimum V［node］in V

11.if P［key］is not partition

12.P.add(key，node)

13.V［node］+=TK［key］

14.remove the key form K

15.end if

16.end while

17.return P

3 實驗結果與分析

文中搭建Hadoop集群來驗證算法的有效性。實驗集群由6臺計算機組成，每臺計算機內存2 G，磁盤空間500 G，奔騰處理器。Hadoop版本1.0.0，操作系統為CentOS6.6，JDK1.6。實驗所采用的測試方法為利用Hadoop進行WordCount計算，并分別與默認Hadoop和文獻［6］中的方法進行比較。

實驗結果如圖4所示。從圖4中可以看到，當數據分布均勻時，即傾斜度為0時，Hash劃分的性能是最好的。隨著數據的傾斜度上升，抽樣的方法運行時間上升緩慢，而Hash上升很快。原因在于當數據分布均勻時，Hash劃分可以保證負載均勻，而抽樣的方法增加了抽樣過程的代價，導致運行時間增加，但是代價很小。當傾斜嚴重時，抽樣的劃分使負載均衡，性能受傾斜影響不大。

從圖4中還可以看出，基于數據本地性的抽樣劃分性能比僅簡單抽樣劃分的性能要好。

4 結束語

文中研究了數據傾斜下的負載均衡優化問題，分析了MapReduce中導致節點負載不均的原因，提出了基于數據本地性的抽樣劃分方法。實驗結果表明，與傳統的Hash劃分和只簡單抽樣的劃分相比，文中提出的方法具有更高的效率。

［1］ Dean J，Ghemawat S.MapReduce:simplified data processing on large clusters［J］.Communications of the ACM，2008，51 (1):107-113.

［2］ Kwon Y C，Balazinska M，Howe B，et al.Skewtune:mitigating skew in MapReduce applications［C］//Proceedings of the 2012 ACM SIGMOD international conference on management of data.［s.l.］:ACM，2012:25-36.

［3］ Ibrahim S，Jin H，Lu L，et al.Handling partitioning skew in MapReduce using LEEN［J］.Peer-to-Peer Networking and Applications，2013，6(4):409-424.

［4］ Ramakrishnan S R，Swart G，Urmanov A.Balancing reducer skew in MapReduce workloads using progressive sampling［C］//Proceedings of the third ACM symposium on cloud computing.［s.l.］:ACM，2012.

［5］ Xu Y，Zou P，Qu W，et al.Sampling-based partitioning in MapReduce for skewed data［C］//ChinaGrid annual conference. ［s.l.］:IEEE，2012:1-8.

［6］ 韓 蕾，孫徐湛，吳志川，等.MapReduce上基于抽樣的數據劃分最優化研究［J］.計算機研究與發展，2013，50(S): 77-84.

［7］ 周家帥，王 琦，高 軍.一種基于動態劃分的MapReduce負載均衡方法［J］.計算機研究與發展，2013，50(S):369-377.

［8］ 宛 婉，周國祥.Hadoop平臺的海量數據并行隨機抽樣［J］.計算機工程與應用，2014，50(20):115-118.

［9］ 萬 聰，王翠榮，王 聰，等.MapReduce模型中reduce階段負載均衡分區算法研究［J］.小型微型計算機系統，2015，36(2):240-243.

［10］傅 杰，都志輝.一種周期性MapReduce作業的負載均衡策略［J］.計算機科學，2013，40(3):38-40.

［11］李 喬，鄭 嘯.云計算研究現狀綜述［J］.計算機科學，2011，38(4):32-37.

［12］劉寒梅，韓宏瑩.基于反饋調度的MapReduce負載均衡分區算法研究［J］.信息通信，2015(10):41-42.

［13］李航晨，秦小麟，沈 堯.數據本地性感知的MapReduce負載均衡策略［J］.計算機科學，2015，42(10):50-56.

Research on Handling Data Skew in MapReduce

WANG Gang，LI Sheng-en
(School of Computer Science and Technology，Shandong Jianzhu University，Jinan 250101，China)

With the rapid development of mobile Internet and the Internet of Things，the data size explosively grows，and people have been in the era of big data.As a distributed computing framework，MapReduce has the ability of processing massive data and becomes a focus in big data.But the performance of MapReduce depends on the distribution of data.The Hash partition function defaulted by MapReduce can’t guarantee load balancing when data is skewed.The time of job is affected by the node which has more data to process.In order to solve the problem，sampling is used.It does a MapReduce job to sample before dealing with user’s job in this paper.After learning the distribution of key，load balance of data partition is achieved using data locality.The example of WordCount is tested in experimental platform.Results show that data partition using sample is better than Hash partition，and taking data locality is much better than that using sample but no data locality.

big data;MapReduce;load balancing;sampling

TP301

A

1673-629X(2016)09-0201-04

10.3969/j.issn.1673-629X.2016.09.045

2015-10-22

2016-02-24< class="emphasis_bold">網絡出版時間:

時間:2016-08-01

國家自然科學基金資助項目(61170052)

王 剛(1990-)，男，碩士研究生，CCF會員，研究方向為大數據、數據庫;李盛恩，教授，研究方向為數據庫、數據挖掘。

http://www.cnki.net/kcms/detail/61.1450.TP.20160801.0842.012.html