王 剛，李盛恩

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

MapReduce中數(shù)據(jù)傾斜解決方法的研究

王 剛，李盛恩

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

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

大數(shù)據(jù);MapReduce;負載均衡;抽樣

0 引言

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

性能優(yōu)化的重點就是負載均衡。在MapReduce分布式計算框架下，用戶提交的作業(yè)被劃分成若干個塊，每個塊被分配給一個Mapper執(zhí)行，Map階段產(chǎn)生的中間結(jié)果經(jīng)過劃分函數(shù)交給Reducer執(zhí)行并產(chǎn)生最終的結(jié)果。整個作業(yè)的完成時間由Reducer運行最慢的決定。當(dāng)節(jié)點的負載出現(xiàn)不均衡時，負載重的節(jié)點會制約作業(yè)的完成時間。因此，每個節(jié)點能否在一致的時間內(nèi)完成是影響分布式計算性能的關(guān)鍵因素。

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

1 相關(guān)工作

1.1 MapReduce

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

Hadoop開源實現(xiàn)了Google的MapReduce模型，并且提供了分布式文件系統(tǒng)HDFS。用戶提交作業(yè)后，數(shù)據(jù)被等大小切分給Map節(jié)點處理，Map節(jié)點執(zhí)行Map函數(shù)產(chǎn)生中間結(jié)果并根據(jù)劃分函數(shù)保存在本地磁盤。Reduce節(jié)點讀取中間結(jié)果執(zhí)行Reduce函數(shù)產(chǎn)生最終的結(jié)果。在這個過程中，用戶不必關(guān)注分布式處理的細節(jié)，作業(yè)調(diào)度、數(shù)據(jù)劃分以及容錯處理這些細節(jié)由MapReduce自動完成。除了Google的內(nèi)部實現(xiàn)外，MapReduce還有一個應(yīng)用廣泛的開源實現(xiàn)Hadoop。

1.2 已有工作

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

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

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

2 基于抽樣的數(shù)據(jù)劃分

2.1 數(shù)據(jù)傾斜

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

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

圖1 哈希劃分不平衡示例

2.2 抽 樣

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

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

2.3 負載均衡方法

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

步驟1:運行一個mapreduce job獲得中間數(shù)據(jù)集樣本，然后統(tǒng)計key的分布，根據(jù)樣本分布產(chǎn)生劃分。劃分結(jié)果用一個映射數(shù)據(jù)結(jié)構(gòu)表示:(k，p)鍵為k被劃分到了p。

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

然而，在MapReduce現(xiàn)有的調(diào)度策略中并未充分考慮數(shù)據(jù)本地性［13］，在任務(wù)的調(diào)度過程中只是簡單地從隊列中取出第一個待分配的任務(wù)給當(dāng)前可用節(jié)點而忽略了中間數(shù)據(jù)的分布特點，因此可能導(dǎo)致大量的中間文件必須跨網(wǎng)絡(luò)傳輸?shù)皆摴?jié)點，如圖2所示。

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

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

結(jié)合抽樣技術(shù)和數(shù)據(jù)本地性算法的具體執(zhí)行過程如下:

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

Step2:統(tǒng)計所有節(jié)點上鍵為key的總數(shù)，用 TK表示。

Step3:將中間結(jié)果按數(shù)量從大到小排序。

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

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

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

圖1的例子利用文中的負載均衡算法進行數(shù)據(jù)劃分之后的結(jié)果如圖3所示?？梢钥闯?，文中所提的利用數(shù)據(jù)本地性的抽樣方法獲得了較好的Reduce端負載均衡，優(yōu)化了默認的哈希劃分和簡單的抽樣劃分。

算法1:基于數(shù)據(jù)本地性的抽樣劃分。

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 實驗結(jié)果與分析

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

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

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

4 結(jié)束語

文中研究了數(shù)據(jù)傾斜下的負載均衡優(yōu)化問題，分析了MapReduce中導(dǎo)致節(jié)點負載不均的原因，提出了基于數(shù)據(jù)本地性的抽樣劃分方法。實驗結(jié)果表明，與傳統(tǒng)的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上基于抽樣的數(shù)據(jù)劃分最優(yōu)化研究［J］.計算機研究與發(fā)展，2013，50(S): 77-84.

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

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

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

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

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

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

［13］李航晨，秦小麟，沈 堯.數(shù)據(jù)本地性感知的MapReduce負載均衡策略［J］.計算機科學(xué)，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">網(wǎng)絡(luò)出版時間:

時間:2016-08-01

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

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

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