利用布隆濾波二次拆分的數據傾斜處理算法

張 強，張學文

(1.重慶大學城市科技學院 電氣信息學院，重慶 402167；2.北華大學 機械工程學院，吉林 吉林 132013)

0 引 言

大數據處理往往采用MapReduce并行任務處理的方式來進行[1]，主要包含映射(Map)操作和歸約(Reduce)操作[2]。其中，Reduce操作中混洗(Shuffle)階段的功能為通過調用分區程序將映射器(Mapper)的輸出結果轉移到歸約器(Reducer)中進行排序(Sort)與歸約(Reduce)操作[3]。分區程序往往采用Hash函數，并忽略Mapper與Reducer之間網絡拓撲以及Mapper輸出數據量規模。缺點在于由于并行計算任務量、數據集物理屬性參數分布特性、Mapper數據規模以及網絡拓撲分布的差異性，Shuffle的時間開銷將變得不可控[4-6]，即數據集的傾斜特性將影響Map-Reduce的計算性能。

文獻[7]以文件均衡偏差值作為數據傾斜的度量標準，設計了面向數據均衡的貪心策略,但缺點在于抽樣統計操作會引入額外的時間開銷。文獻[8]通過JobTracker監控Map階段產生的數據規模，提出改進的Hash分區函數將較大數據規模的任務索引到負載較輕的Reducer上。此種方案同樣引入額外的時間開銷，且降低了Mapper與Reducer的并發度。此外，現有研究仍有以下兩個缺點：①現有文獻大多以Reducer同構為假設前提設計數據傾斜消除策略，但忽略了Reducer硬件配置、計算資源等差異性對MapReduce性能的影響；②構建Mapper與Reducer最優映射關系為NP難問題[9]，需在時間開銷與映射方案選擇間折中選取。

針對以上不足，提出一種基于流量開銷自適應感知和自動學習機的數據傾斜處理算法。主要創新點為：

(1)在Map階段檢測過載Reducer，并考慮到硬件配置、計算資源等異構特性的實際約束，提出二次分拆機制對Mapper輸出數據集進行分片，設計了基于布隆濾波的Hash分區算法，使Mapper根據自身流量開銷部署到Reducer。與傳統算法相比，所提策略能夠根據Mapper流量開銷自適應選擇Reducer的候選范圍，縮短Reducer選擇的時間開銷。

(2)在Reduce階段設計了基于學習自動機的分區算法，使用均勻分布抽樣算法獲取任務類型與規模，并以Reducer候選范圍為約束條件搜索Mapper與Reducer在有限時間內的近似最優映射關系。與傳統方案相比，學習自動機方法有效解決了映射構建時的NP難問題，提升分區算法執行效率。

1 基于布隆濾波Hash分區的流量開銷感知機制

Mapper中的輸出為具有的鍵值對。傳統Hash分區函數中，具有相同Hash值的鍵值對存儲為同一個分區[10]。然而，此種方法忽略了Reducer的異構特性且難以實現對數據均勻劃分。其原因在于key的頻率與數據量分布存在固有的不均分分布特性。因此，提出一種基于布隆濾波的改進Hash分區算法，自適應感知各Mapper的數據流量開銷，對Mapper內數據進行二次拆分，其基本示意圖如圖1所示。

圖1 所提改進Hash分區策略

1.1 提前傳輸與二次拆分機制

為縮短MapReduce計算時間開銷，設Mapper執行完成度達到某一閾值后，對于已完成的任務進行提前傳輸。設θ為Mapper完成任務量占總任務量的百分比，即

(1)

式中：Taskfinished和Task∑分別表示已完成的任務數量以及總的任務數量。符合傳輸條件的判據為數據量分片值大于分片均值，分片均值計算如下

(2)

式中：Saverage為分片均值，Num為Reducer的總數量。

在Map階段完成后，主節點統計所有中間結果的數據傾斜程度，并按照式(2)重新計算現階段的負載均值。對于超過負載均值的數據進行二次拆分，其拆分公式為

(3)

采用式(3)所示拆分操作后，盡管主節點中保留了二次拆分的具體列表，但Mapper端的中間結果中，分片仍然是完成的臨時文件。因此，在確定各子分片與Reducer間的對應關系后，臨時文件需要整體打包發送至Reducer節點。若所有分片無差別地傳輸至Reducer勢必引發網絡負載的激增。因此，設計了布隆濾波器以選擇僅符合條件的子分片數據進行傳輸。

1.2 布隆濾波傳輸機制

如圖2所示，對于每一個子分片，首先創建具有n個特征比特串和k個Hash散列函數的布隆濾波器。傳輸過程包含如下兩個子過程。

圖2 二次拆原理

(1)特征置位階段：初始條件n個特征比特串全為0。進一步地，定義子分片的Key值為每個散列Hash函數的輸入，輸出則為[0,n-1]之間的整數，從而映射到特征比特串的n個位置。若對應的特征位置為0，則將其置為1；反之，若該位置已經為1，則保持不變。

(2)查詢傳輸階段：對特征比特串1～n通過Hash函數進行逐次計算，生成對應的k個Hash值，并以此作為布隆濾波閾值。可以嚴格證明，布隆濾波的錯誤傳輸概率P滿足如下關系[11]

(4)

式中：Numbit為通過布隆濾波器的特征比特串。

基于上述分析可知，與傳統一次分區操作相比，所設計方法實現了對規模過大的數據塊的自適應二次拆分，且避免了數據塊重復傳輸帶來的額外網絡開銷。

2 基于學習自動機的Mapper-Reducer最優映射機制

2.1 變動作學習自動機基本原理

對于具有r個Mapper和s個Reducer的MapReduce框架，其可能的映射關系為sr種，當r和s數量較大時，搜索并建立Mapper與Reducer最佳映射關系可視為NP難問題。本文選擇可變動作學習自動機進行最優映射關系構建。基本原理為：設B={b1,b2,…,bn}為一個有限動作集，且這組動作可以隨時間的變化而變化。自動機可以從中選擇動作，若在時間t暫時禁用某些動作，從而得到B′={b′1,b′2,…,b′n′}， 且n′

學習自動機中規定，所有動作被激活的概率之和始終等于1[12]，即

(5)

學習自動機算法中，活動動作總是基于其概率向量隨機選擇，并且基于環境響應，更新活動動作的概率。在更新活動動作的概率之后，自動機必須再次激活已被禁用的所有動作并更新概率向量，使所有行動的總概率等于1，更新公式如下

(6)

2.2 算法流程

由于第1節中已然對Mapper的數據量進行了平衡處理，而Reducer的工作能力未知，顯然若Reducer為同構的，則最佳的映射關系下所有的Reducer工作負載應相等；反之，若Reducer為異構的，則每個Reducer上的工作負載應與其所提供的處理能力成正比。

所提學習自動機的輸入包含3個內容：①Reducer的數量；②每個Mapper的平均數據量，即Saverage；③Reducer可提供的計算資源。相應的，其輸出則為Mapper到Reducer間的映射列表。不妨令學習自動機的集合定義為{LA1,LA2,…,LAr}，每個學習自動機的動作集合為A={A1,A2,…,As}，表示所選擇的Reducer集合。每個自動機中，選擇每個Reducer的概率記為p={p1,p2,…,ps}。所有自動機過程開始時，選擇每個Reducer的概率是相同的且等于1/s。進一步地，令CS={CS1,CS2,…,CSr}表示Mapper上處理的數據集，RC={RC1,RC2,…,RCs}表示每個Reducer可提供的處理性能。因此，基于學習自動機的映射構建機制如下所示。

不失一般性，設Reducer數量為3，Reducer1、Redu-cer2、Reducer3上的期望負載比例為1∶1∶3，則RC可設為{RC1=1,RC2=1,RC3=3}。此種情形下，若要在3個Reducer上分配1200個Mapper的輸出數據量，則可根據下式進行分配，即

(7)

式中：Porj為放置在第j個Reducer上的數量，WL={WL1,WL2,…,WLs}為Mapper的數據量大小。從而Reducer1、Reducer2、Reducer3的數據量為

(8)

(9)

(10)

如前所述，學習自動機的輸出為一個具有s個成員的列表集合，不妨記為List={List1,List2,…,Lists}，Listj表示發送到第j個Reducer。學習自動機算法在Map操作結束后迭代進行直至找到近似最優映射方案或達到迭代終止的閾值。

考慮到Reducer的計算能力差異性，定義如下的負荷變異系數對Reducer整體的工作負載進行度量，即

(11)

式中：STD(·)、AVG(·)分別為標準差和均值操作，workload為Reducer的負載集合。在同構Reducer環境下，最優映射建立時每個Reducer的工作負載相同，從而工作負荷變異系數為0；而異構Reducer環境下，最優映射關系下的COV應接近于0。

因此，所提迭代學習自動機流程如下：

(1)學習自動機激活：CS集合中的每個成員進行迭代次數為r的循環迭代，學習自動機執行如下操作：當負載大于其未動作的部分時，激活對應的自動機動作并按式(6)進行概率更新。而每個Reducer分配的任務量則根據式(7)進行分配。

(2)Reducer映射確立：學習自動機根據其概率相量隨機選擇Reducer，例如Ae，即Reducere與CSi建立映射關系。從而將CSi的數據規模添加到workload集合中，并更新Reducer整體的數據規模。對每個CS重復上述過程。運行循環r次后，每個群集將被放置在Reducer上。

(3)映射方案評估，將新發現的解決方案的COV與最佳解決方案的COVbest進行比較。如果新解決方案的COV小于最佳解決方案的COV，則自動機的所有選定動作都會受到獎勵，否則將受到處罰。每次迭代完成后具有最小的COV的映射方案保存為新的最優解決方案，并在下一次迭代過程中重新激活每個自動機的所有禁用動作，并根據等式(6)更新其概率向量。此過程繼續，直到達到接近最優的解決方案或預定義的閾值。

算法最終返回的為有限迭代步驟下的最優解決方案。其中，時間復雜度為O(itrmax×r×s)，其中itrmax為最大迭代次數。所提算法的步驟如下所示。

算法1：基于學習自動機的分區算法

輸入：CS={CS1,CS2, …,CSr}，RC={RC1,RC2, …,RCs}，r

輸出：List={List1,List2, …,Lists}為s列集合，初始值為?。

假設：

(1){LA1,LA2, …,LAr}為r個自動學習機，其中r是Mapper節點數量，每個自動學習機具有s個動作，s是Reducer數量；

(2)令s個動作的概率矢量在所有的自動機中都初始化為同樣的值1/s；

(3)令Itrmax為最大迭代次數，ε為接近于0的值。

算法流程：

步驟1 初始化：itr=0;COV=∞;Sizeworkload=0//當前Reducer節點總負載大小。根據式(11)計算當前初始最優COVbest

步驟2 while(itrε)

Begin Forj=1 tordo

Begin

Else 隨機選擇一個LAj的活動動作，如Ae，并根據式(5)更新LAj的概率矢量

workloade+=CSe×Sizee;

Sizeworkload+=CSe×Sizee;

Liste=Liste∪CSe。

End

根據式(11)計算COV

If (COV

保存List集合作為新的可行解,COVbest=COV, flag=0;

Else flag=1

End if

Forj= 1 tordo

Begin

If flag==0，獎勵選定動作LAj

Else 懲罰選定動作LAj

End if

啟用所有禁用的動作并根據式(11)重構概率矢量

End

itr++;

END While

輸出List={List1,List2, …,Lists}

3 實驗驗證與結果分析

3.1 實驗環境設置

為驗證所提方案的可行性與優異性，本文在Hadoop 2.7.1環境下進行實驗驗證。其中，5臺虛擬機配置與惠普 HPE ProLiant DL20Gen10服務器上，并采用KVM作為虛擬機管理程序，表1示出了虛擬機的配置參數。

表1 虛擬機參數配置

本節中，除特殊說明外所有實驗均采用Hadoop中默認配置參數。此外，在所有實驗中使用參數α=0.02的線性學習方案，并使用針對標準數據集的常用操作——文本數據集字數統計來分析所提數據傾斜處理算法的性能。其中，文本中的單詞出現頻率符合如式(12)所示的zeta分布，σ的變化范圍為[0.5,3]。所有算法均運行20次并計算其平均值進行對比分析

(12)

本文選擇如下算法作為對照實驗：

(1)傳統Hash算法[13]：在Hadoop和Spark中用于數據分區的默認算法；

(2)Range算法[14]：該算法分區結果為可變范圍，并保留了Reducer分區順序；

(3)C2 WC算法[15]：該算法通過對MapReduce任務量進行采樣并使用啟發式方法進行Mapper任務量聚類組合；

(4)SkewTune算法[16]：該算法在Mapper執行期間在檢測落后計算節點的基礎上在其它節點間分配剩余工作負載。

3.2 評估指標

為了比較各算法的優劣性，采用如下指標：

(1)作業運行時間。此參數等于作業完成時間減去作業開始時間；

(2)負載變化系數。衡量Reducer節點間負載均衡度，根據式(11)計算；

(3)負載極值比率。通過將Reducer節點上的最小工作負載除以最大工作負載得到。

3.3 性能對比與分析

首先按照式(12)生成大小為分別為1 GB，3 GB和5 GB的3組數據集，數據集的傾斜程度與σ成正相關關系。圖3示出了σ=0.5和σ=3，數據量從1 GB到5 GB的作業運行時間。可以看出，在σ取值較小和σ取值較大兩種場景下，5種算法隨數據量變化的趨勢相接近。在數據量較低時，5種算法的運行時長較為接近，隨著數據量的增加，算法的運行時間均出現了明顯的上升，而所提算法的運行時間比其它算法的上升幅度緩慢2.4%和15.8%以上。此外，顯然在任何數據量規模下作業運行時間都小于其它算法的作業運行時間。這是由于本文所提算法采用提前傳輸機制與二次拆分方法，使已完成任務的Mapper節點在不等待其余節點完全工作結束后提前傳輸，并將規模較大的計算負載分攤至其它節點，從而節省了Mapper與Reducer間的傳輸時間以及Mapper的計算時間。因此，所提出的算法在處理數據傾斜問題中能具有更快的處理速度。

圖3 不同算法下作業運行時間

進一步地，采用負載變化系數以及負載極值比率來分析不同算法下Reducer上工作負載的分布均衡度。如果COV的值接近于零，則意味著存在更平等的分布。而負載極值比率則反映了Reducer上的最小工作負載大小相對于分布在它們上的最大工作負載大小。該值越接近1，則最小工作負載與最大工作負載之間的距離越短，因此意味著發生更公平的分配。

圖4(a)和圖4(b)分別示出了不同數據集規模與不同zeta分布參數下Reducer節點的負載變化系數與負載極值比率，從上至下所用算法分別為：所提算法、傳統Hash算法[13]、Range算法[14]、C2 WC算法[15]和SkewTune算法[16]。由圖可知，所有算法中的負載變化系數與zeta分布參數σ和數據集規模大小成正相關性，而負載極值比例則與上述兩個參數成負相關性。然而，當zeta分布參數固定，即數據傾斜程度相同時，所提算法相較于Hash算法、Range算法、C2 WC算法和SkewTune算法的負荷變化系數平均小31.96%、35.88%、44.36%、54.12%、62.01%；當數據集規模固定，即MapReduce所需處理的任務量固定時，所提算法相較于Hash算法、Range算法C2 WC算法、SkewTune算法的負荷變化系數平均小28%、33.3%、40%、48.57%和55%。類似的，zeta分布參數固定時，所提算法相較于Hash算法、Range算法C2 WC算法、SkewTune算法的負載極值比率大8.07%、9.72%、14.4%、22.91%和34.74%，而數據集規模固定時，所提算法相較于Hash算法[13]、Range算法[14]、C2 WC算法[15]和SkewTune算法[16]的負載極值比率大7.317%、8.974%、7.5%和8.07%。結合作業運行時間實驗結果，表明所提算法在更短的運行時間內能夠獲取比其它傳統算更優的執行性能。

圖4 不同數據集規模與不同zeta參數下負載變化系數與負載極值比率

4 結束語

針對MapReduce平臺下數據傾斜特性帶來的Reduce階段執行效率下降的問題，提出了基于布隆濾波二次拆分與學習自動機最優映射構建的數據傾斜處理算法。由于算法中采取已完成子任務提前傳輸機制、過載Reducer節點任務分攤機制與自動學習機迭代尋優機制，與傳統數據傾斜處理算法相比，所提算法使Reducer節點能夠在更短的工作運行時間里獲得更小的負荷變化系數與更大的負荷機制比率，從而有效均衡了Reducer節點間的處理負荷。

由于本文假設數據集是真實可信的，因此，未來工作將著眼于數據真實性存疑時所提算法運行性能的提升研究。