基于MapReduce的等值連接中數(shù)據(jù)傾斜問題研究

褚龍現(xiàn)

摘要：針對MapReduce計算框架下實(shí)現(xiàn)數(shù)據(jù)表等值連接時不能很好地處理數(shù)據(jù)傾斜的問題，詳細(xì)分析了數(shù)據(jù)傾斜帶來的任務(wù)負(fù)載不均勻問題和解決思路，結(jié)合兩表之間傳統(tǒng)連接算法和廣播連接算法思想，提出將傾斜數(shù)據(jù)和非傾斜數(shù)據(jù)區(qū)別對待的分區(qū)連接算法。實(shí)驗(yàn)結(jié)果表明，提出的算法很好地解決了數(shù)據(jù)傾斜問題下任務(wù)負(fù)載均衡問題，有效提高了兩表之間等值連接查詢效率。

關(guān)鍵詞：數(shù)據(jù)傾斜；連接；MapReduce；分區(qū)

中圖分類號：TP311 文獻(xiàn)標(biāo)識碼：A 文章編號：1009-3044（2016）32-0226-03

Research on Data Skew in Equal-join based on MapReduce

CHU Long-xian

（Computer School， Pingdingshan University， Pingdingshan 467000， China）

Abstract： Aiming at the problem that data skew cant be handled well when data table is joined by MapReduce， this paper analyzes the problem of task load unevenness and solution in detail. Combining the traditional join algorithm between two tables and broadcast join algorithm， we propose a partitioning algorithm that treats skewed and non-skewed data differently. The experimental results show that the proposed algorithm solves the problem of task load balancing under data skewing and improves the query efficiency of equal-join between two tables.

Key words： Data skew； join；MapReduce；partition

1 引言

計算機(jī)網(wǎng)絡(luò)技術(shù)的不斷發(fā)展和社會信息化建設(shè)的不斷加強(qiáng)促使數(shù)據(jù)規(guī)模的快速增長，PB級大小的數(shù)據(jù)管理在很多行業(yè)成為常態(tài)。隨著大數(shù)據(jù)的出現(xiàn)，數(shù)據(jù)處理必然需要分布式計算。MapReduce是Hadoop平臺的并行計算框架，可以實(shí)現(xiàn)對存儲在HDFS（Hadoop Distributed File System）中的大數(shù)據(jù)進(jìn)行分布式處理，提高大數(shù)據(jù)處理效率[1，2]。在網(wǎng)絡(luò)應(yīng)用中，搜索和數(shù)據(jù)庫查詢領(lǐng)域離不開連接操作[3]，所以研究MapReduce下數(shù)據(jù)連接算法的優(yōu)化有著重要意義。

MapReduce提供了數(shù)據(jù)連接并行處理的基本算法，但是在同一個數(shù)據(jù)集中經(jīng)常出現(xiàn)某些數(shù)值大量出現(xiàn)，導(dǎo)致數(shù)據(jù)分布不平均問題，利用現(xiàn)有算法將導(dǎo)致節(jié)點(diǎn)負(fù)載不均衡[4，5]。本文以兩個數(shù)據(jù)表連接為目標(biāo)，研究數(shù)據(jù)傾斜情況下大數(shù)據(jù)連接操作的算法優(yōu)化問題，提出分區(qū)連接算法，提高連接執(zhí)行效率。

2 相關(guān)工作

2.1 MapReduce

MapReduce是基于Hadoop平臺的并行計算框架，使用map和reduce函數(shù)實(shí)現(xiàn)數(shù)據(jù)的分布式處理，map函數(shù)對分片數(shù)據(jù)執(zhí)行讀取、分區(qū)、排序和合并后提交到reduce執(zhí)行[6]。在執(zhí)行任務(wù)過程中，由master節(jié)點(diǎn)負(fù)責(zé)系統(tǒng)控制和任務(wù)分配等工作，slave節(jié)點(diǎn)執(zhí)行具體任務(wù)。MapReduce工作流程如圖1所示。

圖1 MapReduce工作流程

2.2 連接查詢

根據(jù)連接運(yùn)算執(zhí)行的時機(jī)，Hadoop下兩表連接主要有分為map端連接和reduce端連接兩種[7]。

map端連接適用于參與連接的其中一個表大小可以緩存到內(nèi)存中，廣播連接是常見的一種算法實(shí)現(xiàn)，主要通過將小表廣播到所有map節(jié)點(diǎn)，然后與每個節(jié)點(diǎn)中存儲的另一個表的數(shù)據(jù)塊進(jìn)行連接，將結(jié)果寫入HDFS中。

reduce端連接可以是標(biāo)準(zhǔn)Hash連接也可以是先通過半連接預(yù)處理數(shù)據(jù)后連接。Hash連接主要在map函數(shù)將標(biāo)記有數(shù)據(jù)來源的元組按照連接屬性進(jìn)行Hash劃分，完成shuffle后，相同連接屬性的兩個表數(shù)據(jù)會劃分到同一個reduce中，在reduce端完成連接操作。

半連接應(yīng)用于大表和小表連接的場景，通過小表中的連接屬性對大表參與連接的數(shù)據(jù)進(jìn)行預(yù)處理過濾，以此減少參與連接的大表數(shù)據(jù)量[8]。在MapReduce中應(yīng)用半連接算法需要三輪map和reduce過程，分別完成小表連接屬性獲取、大表數(shù)據(jù)過濾和最后reduce的連接。

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

數(shù)據(jù)傾斜主要描述的是數(shù)據(jù)表中某個特定值出現(xiàn)的頻率遠(yuǎn)遠(yuǎn)大于其他值，在分布式存儲情況下，參與連接運(yùn)算的數(shù)據(jù)表數(shù)據(jù)傾斜將會對查詢執(zhí)行效率產(chǎn)生巨大影響。因?yàn)閿?shù)據(jù)源數(shù)據(jù)傾斜導(dǎo)致在MapReduce計算框架中，默認(rèn)的分區(qū)策略使reduce各任務(wù)接受的數(shù)據(jù)量可能不均衡，從而出現(xiàn)負(fù)載重的reduce一直處于工作狀態(tài)，整個任務(wù)完成時間大大增加[9]。

3 負(fù)載均衡連接算法

3.1負(fù)載均衡處理方案

reduce任務(wù)負(fù)載不均衡主要由于MapReduce簡單地對連接屬性（key）進(jìn)行Hash導(dǎo)致，為此可以優(yōu)化Hash函數(shù)，將key值按照區(qū)間進(jìn)行劃分，相同區(qū)間的數(shù)據(jù)分區(qū)到同一個reduce中，最終使得reduce任務(wù)負(fù)載趨于均衡。實(shí)際應(yīng)用中主要有兩種區(qū)間分區(qū)，一種是簡單區(qū)間分區(qū)，一種是虛節(jié)點(diǎn)分區(qū)[10]。

3.1.1區(qū)間分區(qū)

（1）簡單區(qū)間分區(qū)

設(shè)定reduce的數(shù)量為n，區(qū)間分割點(diǎn)集為{r1，r2，…，rn-1}，所有數(shù)據(jù)按照key在分割點(diǎn)集中的位置分為n個部分，則自定義Hash函數(shù)可以將數(shù)據(jù)劃分為n的區(qū)間段，每個reduce處理一個區(qū)間段中的數(shù)據(jù)，以此實(shí)現(xiàn)reduce任務(wù)的均衡。

（2）虛擬節(jié)點(diǎn)分區(qū)

設(shè)定reduce的數(shù)量為n，區(qū)間分割點(diǎn)集為{r1，r2，…，rk*n-1}（k=1，2，…，m），所有數(shù)據(jù)按照key在分割點(diǎn)集中的位置分為k*n個部分，從k*n個區(qū)間段中依次取出n個執(zhí)行Hash函數(shù)重分區(qū)到reduce中，直至全部取完，每個reduce處理一個區(qū)間段中的數(shù)據(jù)，以此實(shí)現(xiàn)reduce任務(wù)的均衡。

3.1.2區(qū)間分區(qū)實(shí)現(xiàn)

區(qū)間分區(qū)的實(shí)現(xiàn)主要是分割點(diǎn)集的確定，對于大數(shù)據(jù)集合，首先進(jìn)行隨機(jī)采樣得到小樣本，再進(jìn)行排序并選取區(qū)間分割點(diǎn)集，最終在一次MapReduce任務(wù)中完成分割。分割點(diǎn)集確定過程如下：

第一步：對連接key進(jìn)行采樣，利用map函數(shù)對key按照比例進(jìn)行過濾，將過濾后的樣本key發(fā)送到同一個reduce中；

第二步：在reduce中得到key排序后的結(jié)果集合K，長度為L，則區(qū)間分割點(diǎn)集合為

R={ri|ri=kj， 1≤i≤n-1， j=1+i*（L/n），kj∈K }

其中，n為連接操作設(shè)定的reduce的數(shù)量。

3.2 分區(qū)連接算法

當(dāng)待分割數(shù)據(jù)中某個值數(shù)據(jù)量特別大時會導(dǎo)致該數(shù)值橫跨多個數(shù)據(jù)區(qū)間，最終大量相同數(shù)值的數(shù)據(jù)分區(qū)到同一個reduce，影響性能。為此，針對數(shù)據(jù)是否傾斜提出不同解決方案。

設(shè)定兩個關(guān)系分別為R（M，N）和S（N，P），其中N是連接key。定義兩個集合Lr和Ls，Lr包含R.N中高頻數(shù)據(jù)，Ls包含S.N中高頻數(shù)據(jù)。利用新的分區(qū)算法完成兩個表的連接操作算法如下：

步驟1：R（M，N）和S（N，P）隨機(jī)分區(qū)到n個reduce中，每個reduce中數(shù)據(jù)為Ri和Si，其中1≤i≤n；

步驟2：在第i個reduce中將Ri分解為三個集合，分別為：

Ri- loc ={r|r∈Ri， Ri.N∈Lr}，Ri-bro={r|r∈Ri， Ri.N∈Ls}，Ri-hash={r|r∈Ri- （Ri-local ∪Ri-bro） }

步驟3：將每個reduce中的Ri-bro集合發(fā)送到所有reduce中，將Ri-hash集合Hash到指定的reduce中，最終每個reduce上得到三個集合：

Ri-local= Ri-loc，Ri-broadcast =∪1≤j≤nRj-bro，Ri-other={r| h（r.N）=i，r∈∪1≤j≤nRj- hash }

步驟4：在每個reduce對S執(zhí)行步驟2和步驟3

步驟5：在每個reduce節(jié)點(diǎn)執(zhí)行

Ri-local∞Si-broadcast∪Si-local∞Ri-broadcast∪Ri-other∞Si-other

步驟6：將結(jié)果寫入HDFS中。

4 實(shí)驗(yàn)

實(shí)驗(yàn)在云平臺上虛擬4個節(jié)點(diǎn)組成Hadoop集群，主節(jié)點(diǎn)一個，從節(jié)點(diǎn)三個。每個節(jié)點(diǎn)為2.6GHZ的雙核CPU，8GB內(nèi)存，64位的CentOS 6.6操作系統(tǒng)，Hadoop版本為2.6.0。

實(shí)驗(yàn)數(shù)據(jù)集使用TPC-H測試集工具生成，測試customer和orders兩個數(shù)據(jù)表的并行連接操作，設(shè)定customer數(shù)據(jù)量為2千萬行，orders數(shù)據(jù)傾斜率設(shè)定為0.3和0.6兩種，連接率為100%，數(shù)據(jù)量的選取如表1所示。

reduce端虛擬節(jié)點(diǎn)分區(qū)連接算法為VQ，本文提出的分區(qū)連接算法為PQ，則在兩種傾斜率下執(zhí)行連接操作時間如圖2所示。

圖2中顯示在數(shù)據(jù)傾斜情況下，隨著數(shù)據(jù)量增加PQ算法執(zhí)行時間比VQ算法執(zhí)行時間減少幅度也會增加；隨著數(shù)據(jù)傾斜率的增大，兩種算法執(zhí)行時間會增長，但是PQ算法增加幅度較小。

5 結(jié)束語

本文研究的重點(diǎn)是優(yōu)化分布式數(shù)據(jù)連接算法，致力于解決數(shù)據(jù)傾斜問題對連接算法的影響。針對傳統(tǒng)區(qū)間分區(qū)算法無法處理傾斜數(shù)據(jù)問題，提出將傾斜數(shù)據(jù)和非傾斜數(shù)據(jù)區(qū)分對待，實(shí)驗(yàn)結(jié)果表明，提出的分區(qū)連接算法在大數(shù)據(jù)集連接操作上很好地解決了數(shù)據(jù)傾斜帶來的負(fù)載不均衡問題，提高了數(shù)據(jù)查詢效率。

參考文獻(xiàn)：

[1]Qin X P， Wang H J， Xiao-Yong D U， et al. Big Data Analysis—Competition and Symbiosis of RDBMS and MapReduce[J]. Journal of Software， 2012， 23（1）：32-45.

[2]趙彥榮，王偉平，孟丹，等.基于Hadoop的高效連接查詢處理算法CHMJ[J].軟件學(xué)報，2012， 23（8）：2032-2041.

[3]高宇飛，曹仰杰，陶永才，等.MapReduce計算模型下基于虛擬分區(qū)的數(shù)據(jù)傾斜處理方法[J]. 小型微型計算機(jī)系統(tǒng)， 2015， 36（8）：1706-1710.

[4]Gufler B，Augsten N，Reiser A，et al.Handing data skew in mapReduce[C].Proceedings of the 1 st International Conference on Cloud Computing and Services Science，2010，146：574-583.

[5]Kwon Y C，Ren K，Balazinska M，et al.Managing skew in Hadoop[J].IEEE Data Eng，Bull，2013，36（1）：24-33.

[6]YANG G. The application of MapReduce in the cloud computing[C]// Proceedings of the 2011 2nd International Symposium on Intelligence Information Processing and Trusted Computing. Piscataway： IEEE， 2011： 154-156.

[7]Blanas S，Patel J M，Ercegovac V，et al.A Comparison of Join Algorithms for Log Processing in MapReduce [C].SIGMOD 10，Indianaplis，Indiana，USA，2010：975-986

[8]金健，陳群，趙保學(xué).數(shù)據(jù)傾斜情況下基于MapReduce模型的連接算法研究[J]. 計算機(jī)與現(xiàn)代化， 2013（5）：22-27.

[9] Kwon Y， Balazinska M， Howe B， et al. A Study of Skew in MapReduce Applications[J]. Open Cirrus Summit， 2011.

[10]Atta F， Viglas S D， Niazi S. SAND Join — A skew handling join algorithm for Google's MapReduce framework[M]. IEEE， 2011.