結合節點計算能力的MapReduce負載均衡方法

胡林發,付曉東,2,劉 驪,劉利軍

(1.昆明理工大學 信息工程與自動化學院,昆明 650500;2.昆明理工大學 云南省計算機技術應用重點實驗室,昆明 650500)

0 引 言

隨著互聯網的迅猛發展,尤其是移動互聯網的應用和大數據的普及,數據量迎來爆炸式增長。MapReduce作為一種分布式計算模型,被廣泛應用在大規模數據并行計算過程中[1-2]。目前針對MapReduce有多種實現,如Apache的開源Hadoop框架,將MapReduce和HDFS分布式文件系統進行完美融合,深受工業界和學術界的青睞[3]。

數據均衡劃分是MapReduce框架在Shuffle階段需要解決的一個重要問題。用戶提交的作業(Job)由一系列分別運行在多臺機器上的Map任務(Mapper)和Reduce任務(Reducer)處理完成,作業的完成時間由運行最慢的Reducer決定[4-5]。Hadoop系統默認采用Hash分區方法,即僅根據關鍵字的哈希值和分區個數確定關鍵字的分區,這種劃分方法雖然可以保證每個分區里不同關鍵字種類數大致相等,但每種關鍵字攜帶的數據記錄條數不一定相等,這會使各分區數據總量大小相差懸殊,從而導致各節點負載不均衡問題。研究表明,采用默認的Hash分區方法,超過90%的作業任務出現了Reducer負載不均衡情況,運行時間高出正常任務的22%-38%[6]。先完成任務的節點需要等待滯后節點任務全部完成才能結束當前作業,若中間數據過于集中在某部分Reducer任務節點,先完成任務的節點必須等待其他節點,這個過程會造成集群資源浪費,延長整體作業完成時間,甚至某些節點因資源不足導致任務中斷,使作業無法繼續推進,從而帶來不好的用戶體驗[7-8]。

文獻[9-10]指出,將Mapper產生的中間數據最優地劃分到不同分區,使各分區負載均衡是一個NP-Hard問題。針對MapReduce框架中存在的負載均衡問題,目前已有兩階段分區[11]、多階段分區[12-13]、數據采樣分區[14-15]、延遲分區[16]、遷移分區[17]等方法,這些方法將集群中各節點看作是計算能力相同的節點,但在實際數據處理過程中不同代的硬件環境會使每個節點計算能力不相同[18],且不同計算節點的性能差異會影響整個系統的計算效率[19]。在異構環境中,即使所有分區得到相同規模的數據,也會因節點處理能力不同導致Reduce任務完成時間產生巨大差異,存在先完成任務的節點等待滯后節點的問題,作業執行時間因此會被延長,集群中部分計算資源會被閑置,從而降低了作業處理效率,浪費了計算資源。

本文提出一種結合節點計算能力的劃分方法,即在數據劃分時結合節點計算能力,使各節點數據負載與節點自身的計算能力相匹配,并使大量數據在節點本地處理,降低網絡傳輸時延,從而提升作業的處理效率。本文的主要貢獻包括以下3個方面。

1)提出在異構環境中使用Reservoir算法對Map任務產生的中間數據進行抽樣,記錄樣本中關鍵字的位置和頻次,并以此建立關鍵字分布矩陣。

2)提出一種結合節點計算能力的分區劃分方法。在制定分區計劃時,本文先采用貪心策略對關鍵字進行初步分區,使各關鍵字劃分到其頻次最高的節點對應分區,然后結合節點計算能力并考慮節點位置關系對初步劃分結果進行調整,使各分區負載均衡。

3)設計了一種均衡性衡量方法,該方法綜合考慮了數據量和節點的計算能力值,有利于更加全面地衡量分區結果的均衡性。

1 異構環境中負載均衡問題

在MapReduce處理作業時,作業任務分為Map和Reduce任務。執行任務的函數均由用戶根據業務需求自定義。將集群中節點數記為r,節點集合為N={n1,n2,…,nr},分區集合為P={p1,p2,…,pr},其中pj(j=1,2…,r)為一個分區。為方便集合節點計算能力劃分,這里將pj分區里所有數據作為節點nj上Reduce任務的輸入。Map任務產生的中間數據為鍵值對形式,分區算法會將中間數據按照關鍵字劃分到不同分區。由Reduce任務輸入限制知,相同關鍵字的數據只能被同一個Reduce任務處理,即?i≠j且i,j=1,2,…,r,pi∩pj=?。

分區集合里每個分區的實際數據量大小為S={s1,s2,…,sr},令C={c1,c2,…,cr}表示每個節點的計算能力值,τ(τ>0)為可設定的閾值,當滿足

(1)

時,集群節點負載不均衡。

(2)

將分區劃分方法記為Π(x),則在Shuffle階段關鍵字kt會根據Π(x)計算得到分區pj←Π(kt),j=1,2,…,r,然后關鍵字kt會被劃分到分區pj中。此時,其他節點上關鍵字為kt的數據需要通過網絡傳輸到nj(j=1,2,…,r),則關鍵字kt需要在網絡中傳輸的數據總量為

(3)

關鍵字K={k1,k2,…,kΩ}中所有關鍵字在經過分區方法Π(x)劃分之后,共需在網絡中傳輸的數據總量用VTRS表示,則

(pj=Π(kt),j=1,2,…,r)

(4)

(5)

VTRS值大小取決于VLocality值大小,當VLocality越大時,VTRS越小,需要在網絡中傳輸的數據就會越少。

在執行Reduce任務時,節點nj(j=1,2,…,r)上處理pj分區里的數據,將pj數據總量記為sj,用FoS(factor of skew)值SFoS衡量節點負載的均衡性,計算式為

(6)

因此,本文解決的負載均衡問題為結合節點的計算能力值,尋找一種分區劃分方法Π(x),使SFoS盡可能小,讓各節點負載接近,同時降低網絡開銷,從而提高作業的處理效率。

2 結合節點計算能力的負載均衡方法

針對異構集群環境中負載不均衡問題,本文提出結合節點計算能力的分區方法LBCC(load balancing in MapReduce combined with computing capacity)。在節點加入到集群時,各節點上運行測試程序,執行默認計算任務。將測試數據集的大小記為V,在節點nj上任務完成所需時間記為Tj,則可得出節點nj計算能力值cj=V/Tj,節點計算能力值集合記為C={c1,c2,…,cr}。在搭建Hadoop環境時,利用配置文件core.xml配置節點的所屬機架信息,方便后續利用節點機架信息調整節點負載。

為使各節點負載均衡并降低Shuffle過程中網絡通信開銷,本文在執行用戶提交的計算作業之前,先運行一個抽樣作業進行數據抽樣,并統計樣本數據里關鍵字的位置和頻次分布,由此得到關鍵字分布矩陣M,然后結合M和節點計算能力值,經過位置劃分篩選高低分區以及分區調整等步驟制定分區計劃并將其寫進緩存文件fcache。分區計劃是計算作業分區劃分的依據,使計算作業任務運行時各節點負載均衡,從而提高集群資源的利用率和作業執行效率。

2.1 運行抽樣作業

在抽樣作業Map階段采用Reservoir抽樣算法對數據集進行抽樣,然后在Reduce階段匯總各節點樣本數據,依據樣本里的關鍵字位置和頻次信息建立關鍵字分布矩陣,并根據分布矩陣和節點計算能力信息制定分區計劃。

在抽樣作業Map任務階段,首先初始化一個關鍵字集合KL,再按行讀取數據集并將數據集中的關鍵字逐一添加進KL中,具體過程如算法1所示。

算法1對數據集中關鍵字進行抽樣

輸入:數據集分片β,樣本容量α。

輸出:關鍵字樣本集合KL.

1.KL←?;

2.cnt←0;

3.forlineinβdo

4.k←getKey(line);

5.cnt++;

6.ifKL.size()<αthen

7.KL.add(k);

8.else

9.t←random.nextInt(0,cnt);

10.ift<αthen

11.KL.replace(t,k);

12.end if

13.end if

14.end for

15.outputKL;

當VLocality取最大值時,VTRS取得最小值,需要在網絡中傳輸的數據量最少。顯然,對于任意kt,若pj←Π(kt)且滿足

(7)

時,VLocality可以取最大值。這里先采用貪心方法,逐一將K={k1,k2,…,kΩ}里關鍵字分配到可以使VLocality值取最大值的分區,由此得到初次劃分結果,然后在此基礎上將各分區里關鍵字進行調整,使各分區負載接近分區期望值,具體過程如算法2所示。

算法2制定分區計劃

輸入:二維矩陣M=[mtj]Ω×r,C={c1,c2,…,cr}.

輸出:分區計劃P={p1,p2,…,pr}.

1.pj←?(j=1,2,…,r);

2.fort←1 toΩdo

3.mtj←max{mt1,mt2,…,mtr};

4.j←getNodeIndex(mtj);

5.pj←pj.add(kt);

6.end for

8.PH←?,PL←?;

9.forj←1 tordo

12.ifsj>ejthen

13.PH←PH∪{pj};

14.else

15.PL←PL∪{pj};

16.end if

17.end for

18.forh←1 toPH.size() do

19.pi←PH.get(h);

20.forktinpido

21.pj←getMinNearPartition(PL,pi);

22.pj.add(kt);

24.PL.remove(pj);

25.end if

26.pi.remove(kt);

28.break;

29.end if

30.end for

31.end for

32.outputP={p1,p2,…,pr}.

算法2中,第2—第6行表示依次將關鍵字kt劃分到kt頻次最大的節點對應分區上, 直到所有關鍵字劃分完畢,得到初步劃分結果P={p1,p2,…,pr}。此時,VLocality取最大值,需要在網絡中傳輸的數據量最小。然而,此時并沒有考慮節點負載均衡性,還需要對初步分區計劃進行調整。對于分區pj(j=1,2,…,r),sj為pj分區里數據總量,每個節點負載期望值用ej表示,ej表達式為

(8)

算法2中第7—第17行表示將分區按照實際負載是否高于均衡值劃分為高分區和低分區,若分區sj>ej則將pj加入高分區集合PH,否則加低分區集合PL。逐漸從高分區里移出關鍵字數據,當高分區的實際總數據量低于期望值時則停止移出。第18—第30行表示收集高分區里的關鍵字,并且將從高分區移出的關鍵字逐一分配給PL中的低分區,從而使低分區數據量逐漸接近期望值。若在關鍵字調整的過程中,當某低分區實際數據量高于均衡值時,則將該低分區移出集合PL。

算法2中getMinNearPartition方法是為了在集合PL中尋找離分區pi最近的節點分區,首先在PL中尋找與pi同一機架且分區負載最小的分區,若存在直接返回,不存在則在其他機架上尋找,具體過程如算法3。

算法3getMinNearPartition方法實現

輸入:低分區集合PL,分區pi.

輸出:PL中離pi最近的分區p.

1.p←null;

2.flag←0;

3.forpjinPLdo

4.ifrack(pi)=rack(pj) then

5.ifp!=null&&getLoad(p)>getLoad(pj)

||p=nullthen

6.p←pj;

7.flag←1;

8.end if

9.end if

10.end for

11.ifflag=1 then

12.returnp;

13.end if

14.returnminLoad(PL);

算法3中第3—第10行表示在集合中尋找與分區pi關聯節點ni同一機架的其他節點對應分區,其中,第4行rack方法的作用是獲取分區的機架位置,第5行getLoad方法表示求取分區的實際負載大小,分區pj的負載計算方法可表示為

(9)

算法3中第11—第14行表示如果在PL中找到了合適分區就直接返回,若沒有找到則利用minLoad方法求取整個PL集合中負載最小的分區作為返回結果。

算法2初步劃分過程中,需要遍歷分布矩陣M中所有行,時間復雜度為O(Ω·r)。在分區篩選過程中遍歷分區集合P={p1,p2,…,pr},時間復雜度為O(r)。在分區調整時,需要先將高分區進行排序,這個過程時間復雜度取決于采用的排序算法,本文采用快速排序,所以時間復雜度為O(ΩlogΩ)。在將高分區里部分關鍵字調整到低分區時,需要通過算法3尋找合適分區,最好的情況下只有少量關鍵字需要進行調整,時間復雜度為O(r),而在最壞的情況下,大量關鍵字需要進行調整,此時時間復雜度為O(Ω·r)。綜上,算法2的時間復雜度為O(ΩlogΩ)。

整個抽樣作業的輸出分區計劃為P={p1,p2,…,pr}。為方便計算作業利用分區計劃進行分區,這里將其轉化為以關鍵字kt為鍵、以kt所屬分區編號為值的鍵值對形式,并將其寫入到緩存文件。

2.2 執行計算作業任務

計算作業以全量數據為輸入,并按照制定的計劃進行分區。在Mapper階段讀取緩存文件fcache,并將其轉化為以關鍵字kt為鍵、分區編號為值的鍵值對結構,將其記為F。分區方法Π(kt)首先在F中查找是否存在關鍵字kt,若存在則直接輸出分區pj(j=1,2,…,r),否則按照Hash方法得到pj,分區流程如圖1所示。

圖1 分區劃分流程圖Fig.1 Partition flow chart

由于在抽樣過程中,可能存在少量頻率較小的關鍵字可能沒有被抽樣到,所以在這里使用Hash方法作為輔助方法。任意關鍵字kt根據Π(kt)計算得到分區pj后,將關鍵字kt與其攜帶的數據寫入到pj分區文件中。在計算作業Reduce階段各計算節點分別從Map任務節點拉取屬于本節點的中間數據分區文件,并運行Reduce任務,直至任務運行結束,輸出作業的計算結果。

3 實 驗

本文LBCC方法在分區劃分時考慮各節點計算能力的同時,對網絡傳輸開銷進行了優化。為驗證本文方法效果,采用NoLFA方法[20]、SBaSC方法[21]和DEFH(default Hash)方法對比。NoLFA方法基于LEEN思想并結合了節點計算能力的差異性,適用于異構集群環境,但其與本文方法相比有以下幾點不同:①NoLFA方法直接在計算作業執行過程中通過主節點獲取關鍵字頻次信息,這增加了主節點負擔,降低了集群元數據處理效率,本文使用抽樣作業得到關鍵字頻次分布信息,避免了元數據處理效率降低問題;②分區計劃制定時,NoLFA方法直接按照LEEN方法思想進行處理,而本文方法在做了一次初步劃分之后,對低分區進行調整,可以快速使最低分區總數量達到均衡值,而且本文方法分區均衡性比NoLFA方法更好;③本文方法在調整分區負載過程中同時考慮了節點計算能力和節點位置差異,能更好地適應異構集群環境。SBaSC方法使用了貪心方法思想劃分分區,達到了均衡各節點負載的目的并且提升了作業計算效率,但其將集群中所有節點看作相同的計算能力,忽略了各節點處理能力的差異性。

3.1 數據集

為測試傾斜度對算法性能的影響,實驗采用人工數據集和2個公開數據集。人工數據集是使用程序生成不同傾斜率的數據集[22],實驗時將人工數據集上傳到HDFS系統,并讓其分散在不同節點上進行存儲,每組實驗均基于該數據集執行單詞統計任務。2個公開數據集分別為維基百科數據集[2]和社交網絡數據集LiveJournal[21]。維基百科數據集包含了大量的文本數據信息,實驗時在對該數據集進行預處理后將其作為單詞統計作業的輸入。LiveJournal數據集中包含了大約1億個用戶社交網絡數據,實驗中使用該數據集作為關聯用戶數目統計作業的輸入。

3.2 實驗環境

實驗采用物理節點與虛擬節點結合的方式,模擬異構集群環境中不同計算能力節點環境。每組實驗涉及到的物理機節點配置為I3、8核、16 GByte內存、500 GByte磁盤空間,虛擬機節點在I5機器上搭建,單個虛擬節點分配4核、8 GByte內存、100 GByte磁盤空間,Hadoop版本為2.10,所有節點均采用CentOS 6.9系統,物理機節點和虛擬節點個數分別由實驗需求確定。

3.3 數據傾斜程度對性能影響

通常關鍵字頻次服從Zipfian分布[4],在關鍵字列表K={k1,k2,…,kΩ}中,排在λ(λ=1,2,…,Ω)位置的關鍵字出現頻率f(λ)可以表示為

(10)

(10)式中,z≥0為傾斜程度控制參數,z值越大則表示數據集中關鍵字的頻次分布越集中,當z=0時表示關鍵字頻率相同,即所有關鍵字頻次均勻分布。

分別設置人工生成不同傾斜率數據集傾斜度z=0.2、z=0.4、z=0.6、z=0.8、z=1.0五組實驗,實驗前準備一個關鍵字個數為20 000的單詞列表,各關鍵字根據(10)式得到關鍵字頻率f(λ)。在向輸出文件里寫數據時,f(λ)作為關鍵字kλ寫入的概率,以此生成包含10億單詞的數據文件。

配置不同分區算法并提交單詞統計作業。搭建包含2個物理節點和3個虛擬節點的Hadoop集群環境,通過文件配置使每個節點既是DataNode節點又是NodaManager節點。實驗結果匯總信息如表1、圖2—圖3所示。

表1 在不同傾斜度下的FoS值Tab.1 FoS value at different skew degree

表1展示的是在不同傾斜度數據集作為輸入的情況下各種算法得到的FoS值(表1中,除傾斜度外的數值為實際數值乘以10-5)。不難發現,在各種傾斜度下,本文LBCC方法FoS值最小,即均衡性表現最好,可以使各節點負載更加均衡。DEFH方法在各種傾斜度下FoS值都最大,均衡性最差,這樣會導致集群匯中部分節點負載遠高于其他節點,從而降低作業的執行效率。NoLFA方法FoS值比LBCC方法高,最大可以是LBCC方法的236.2倍,說明此時NoLFA分區結果節點負載均衡程度遠不如本文LBCC方法。

圖2 不同傾斜度下本地化率值Fig.2 Locality value at different skew degree

圖3 不同傾斜度下執行時間Fig.3 Execution time at different skew degree

由圖2可見,在相同數據量下,隨著關鍵字頻次傾斜度的增加,Locality值呈下降趨勢,即需要在網絡中傳輸的數據量逐漸增加。DEFH方法、SBaSC方法變化幅度比較小,因其沒有考慮網絡開銷優化,這2種方法需要在網絡中傳輸的數據量占總數據量20%左右。當傾斜率較高時,LBCC方法Locality值會比NoLFA稍低,這是由于在數據傾斜率較高時部分關鍵字在各節點分布不均勻,本文方法在調整分區過程中,為了使各分區均衡性更好,會結合節點計算能力和節點位置信息將部分關鍵字調整到最低分區,會犧牲一部分Locality值,相對于NoLFA方法在傾斜率較高時Locality值會存在一定差距,但比其他方法本地化率更高。

由圖3可見,在數據總量相同且節點個數一定情況下,作業執行時間也隨之增大。LBCC方法結合節點計算能力將中間數據更加均衡地劃分,縮短了整體作業的完成時間。LBCC方法相較于NoLFA、SBaSC、DEFH方法在效率上都有較大提高。

3.4 節點個數對性能影響

為測試異構環境下節點個數對算法性能的影響,實驗環境初始設置為1個物理節點和2個虛擬節點共3個節點,之后每組實驗在此基礎上依次增加1個物理節點和1個虛擬節點,節點個數依次設置為3、5、7、9、11個。本次實驗分別采用維基百科數據集和社交網絡LiveJournal數據集,每次配置好環境后,將數據集上傳到HDFS文件系統,數據文件會以塊的形式分散存儲在集群中各節點上。實驗中每種算法運行多次后求取各項指標平均值,匯總信息如表2—表3、圖4—圖7所示。

表2 在維基百科數據集上FoS值Tab.2 FoS value for different number of data nodes on Wikipedia dataset

表3 在LiveJournal數據集上FoS值Tab.3 FoS value for different number of data nodes on LiveJournal dataset

由表2—表3可知,在不同節點個數實驗環境中,無論采用哪種數據集作為作業的輸入,本文LBCC方法分區結果的FoS值最低,即均衡性表現最好,可以使各節點負載更加均衡。本文LBCC方法在調整各節點負載時考慮了節點計算能力,讓各分區之間的負載與節點自身計算能力相匹配,與其他各節點負載相均衡。 DEFH方法根據關鍵字哈希值進行劃分,并沒有考慮各節點的負載均衡性,所以FoS值比較大。另外在異構環境中,SBaSC沒有考慮節點的計算能力,所以導致各節點負載差異也很大。

圖4 在維基百科數據集上不同節點個數下的本地化率Fig.4 Locality value for different number of nodes on Wikipedia dataset

圖5 在LiveJournal數據集上不同節點個數下的 本地化率Fig.5 Locality value for different number of nodes on LiveJournal dataset

由圖4—圖5可見,隨著節點的增加,Locality值總體上呈下降趨勢。文獻[5]指出,相同關鍵字的頻次在集群中各節點均勻分布時,數據本地化率取決于節點的個數,即VLocality=1/r,在數據值上將與公式計算的結果相等。由此可知,隨著節點個數的增加,Locality值會隨之下降。在不同節點個數環境下,NoLFA方法和LBCC方法的分區結果中本地化率比較接近,SBaSC和DEFH方法由于沒有考慮網絡開銷優化,Locality值比較低,需要在網絡中傳輸的數據量比較大。數據集LiveJournal上關鍵字的頻次比較集中,大量的關鍵字攜帶的數據可以在節點本地進行處理,不需要通過網絡傳輸到其他節點,所以通過LBCC和NoLFA方法得到的Locality值比較高。

由圖6—圖7可見,隨著節點個數的增加,任務完成時間逐漸降低。另外,在每組實驗中,本文LBCC方法在效率上優于其他分區方法。圖7中NoLFA和LBCC方法相較于圖6差別較大,這是由于使用NoLFA和LBCC方法可以使社交網絡LiveJournal數據集絕大部分在節點本地處理,另外在異構環境中,考慮了節點負載均衡性,使各節點的負載與節點計算能力相匹配。在維基百科數據集上,LBCC方法在作業運行效率上比NoLFA方法提高7.0～15.4百分點,比SBaSC方法提高17.9～23.1百分點,比DEFH方法提高11.0～30.8百分點。在社交網絡LiveJournal數據集上,LBCC方法在效率上比NoLFA方法提高2.8～7.6百分點,比SBaSC方法提高8.1～15.4百分點,比DEFH方法提高10.1～15.9百分點。

圖6 在維基百科數據集上不同數據節點下 任務完成時間Fig.6 Execution time for different nodes on the Wikipedia dataset

圖7 在LiveJournal數據集上不同數據節點下 任務完成時間Fig.7 Execution time for different number of nodes on LiveJournal dataset

4 結束語

本文提出通過Reservoir抽樣方法獲取Map產生的中間數據分布信息,然后結合節點計算能力解決MapReduce在分區過程中的負載均衡問題。實驗結果表明,本文方法得到的分區結果會使各節點負載更為均衡,提高了作業處理效率,同時優化了網絡傳輸代價。本文方法在集群異構的環境中具有良好的性能優勢,計算效率相對于現有分區方法有顯著提升,為MapReduce計算模型負載均衡提供了一種更加高效的解決方案。