基于MapReduce的多元連接優(yōu)化方法

李甜甜 于 戈 郭朝鵬 宋 杰

1（東北大學計算機科學與工程學院 沈陽 110819）2（東北大學軟件學院 沈陽 110819）（litiantian＿neu＠163．com）

基于MapReduce的多元連接優(yōu)化方法

李甜甜1于 戈1郭朝鵬2宋 杰2

1（東北大學計算機科學與工程學院 沈陽 110819）2（東北大學軟件學院 沈陽 110819）（litiantian＿neu＠163．com）

多元連接是數(shù)據(jù)分析最常用的操作之一，MapReduce是廣泛用于大規(guī)模數(shù)據(jù)分析處理的編程模型，它給多元連接優(yōu)化帶來新的挑戰(zhàn)：傳統(tǒng)的優(yōu)化方法不能簡單地適用到MapReduce中；MapReduce連接執(zhí)行算法尚存優(yōu)化空間．針對前者，考慮到I?O代價是連接運算的主要代價，首先以降低I?O代價為目標提出一種啟發(fā)式算法確定多元連接執(zhí)行順序，并在此基礎上進一步優(yōu)化，最后針對MapReduce設計一種并行執(zhí)行策略提高多元連接的整體性能．針對后者，考慮到負載均衡能夠有效減少MapReduce的“木桶效應”，通過任務公平分配算法提高連接內(nèi)部的并行度，并在此基礎上給出Reduce任務個數(shù)的確定方法．最后，通過實驗驗證本文提出的執(zhí)行計劃確定方法以及負載均衡算法的優(yōu)化效果．該研究對大數(shù)據(jù)環(huán)境下MapReduce多元連接的應用具有指導意義，可以優(yōu)化如OLAP分析中的星型連接、社交網(wǎng)絡中社團發(fā)現(xiàn)的鏈式連接等應用的性能．

多元連接；執(zhí)行計劃；I?O代價；性能優(yōu)化；MapReduce編程模型；負載均衡

連接運算根據(jù)連接條件把2個或多個關系中的記錄組合為一個結果數(shù)據(jù)集，包含連接運算的查詢簡稱為連接查詢．連接查詢在數(shù)據(jù)分析中很常見，TPC－H提供的22個查詢用例中有16個涉及到此類查詢［1］．當一個連接查詢涉及n個關系時，稱為n元連接；當n＞2時，稱為多元連接，多元連接是數(shù)據(jù)分析中最常用的操作之一．此外，在當今的大數(shù)據(jù)環(huán)境下，MapReduce［2］編程模型被廣泛用于大規(guī)模數(shù)據(jù)集的分析處理．目前，MapReduce中數(shù)據(jù)分析的優(yōu)化工作包括索引、數(shù)據(jù)布局、查詢優(yōu)化、迭代處理、公平負載分配以及交互式處理等方面［3］．基于此，我們分析MapReduce給多元連接的優(yōu)化帶來的新挑戰(zhàn)．

首先，多元連接查詢依賴良好的執(zhí)行計劃．傳統(tǒng)的執(zhí)行計劃確定方法［4］不滿足MapReduce特性，無法通過簡單的適應性更改應用到現(xiàn)有的優(yōu)化中．另外，現(xiàn)有基于MapReduce的執(zhí)行計劃確定方法［5－6］復雜度較高，應用范圍受限．因此，亟需提出一種新的滿足MapReduce特性且復雜度較低的執(zhí)行計劃確定方法．此外，我們還注意到，無論是傳統(tǒng)研究還是現(xiàn)有研究，其執(zhí)行計劃都只確定了連接的執(zhí)行順序，并未考慮無依賴關系的連接操作間的并行執(zhí)行策略．

其次，良好的執(zhí)行計劃固然重要，對執(zhí)行框架的優(yōu)化同樣可以有效地提高多元連接的性能，這一點在分布式環(huán)境中尤為重要．一種公平的并行任務負載分配方法可以有效地減少MapReduce中的“木桶效應”，從而提高連接操作內(nèi)部的并行度．然而，就我們所知，目前沒有針對連接運算的MapReduce負載均衡方法，且現(xiàn)有的通用方法［7－9］僅考慮了任務的輸入代價，不適用于連接運算，因為它的輸出代價也不可忽略．

本文研究MapReduce環(huán)境下多元連接的優(yōu)化方法，基于上述分析提出以下問題：1）多元連接執(zhí)行計劃的解空間很大，短時間內(nèi)很難找到最優(yōu)解，那么能否通過某個復雜度較小的算法快速找到一個近似最優(yōu)解；2）連接運算屬于I?O密集型運算，I?O為主要代價，那么能否針對MapReduce特性提出I?O代價模型，并選擇代價最小的執(zhí)行計劃；3）連接順序確定后，不存在依賴關系的連接操作可以并行執(zhí)行，那么當存在多個滿足并行執(zhí)行的連接操作時該如何選擇；4）執(zhí)行框架的優(yōu)化中，負載均衡能夠有效地減少“短板效應”，那么此處的連接負載又該如何定義．這些問題的求解存在一定程度的挑戰(zhàn)，就我們目前所知，尚未發(fā)現(xiàn)能夠完全解決上述問題的研究工作．

本文首先通過分析多元連接執(zhí)行計劃解空間的縮減方法、MapReduce連接算法的I?O代價模型、Replicated Join①的優(yōu)劣以及MapReduce作業(yè)的并行執(zhí)行特性，最終確定了執(zhí)行計劃的優(yōu)化方法；接著，結合MapReduce框架分析連接運算的特性，提出負載均衡模型及其對應的均衡算法，并在此基礎上提出Reduce任務個數(shù)的確定方法．大量實驗驗證了本文提出的優(yōu)化方法的有效性．

1 相關工作

現(xiàn)有多元連接的優(yōu)化研究可歸為以下3類：執(zhí)行計劃的優(yōu)化、連接算法的優(yōu)化和執(zhí)行框架的優(yōu)化．

對于第1類研究，文獻［4］將n元連接拆分為n－1個2元連接，每個2元連接對應一個MapReduce作業(yè)（后文如不特殊指明，作業(yè)均為MapReduce作業(yè)），然而該方法針對的是傳統(tǒng)的多處理器計算環(huán)境，不適用于MapReduce，且該方法在n較大時會導致較高的作業(yè)初始化代價以及中間結果的存儲代價．文獻［10］針對鏈式連接提出使用平衡二叉樹的方式來執(zhí)行多元連接，但其并未給出平衡二叉樹的構建規(guī)則．文獻［11］在一個作業(yè)中完成所有的連接運算，然而該方法在n較大時會因為數(shù)據(jù)需要傳輸

①Replicated Join：在一個MRJ中執(zhí)行多個連接操作．此時，同一個Key－Value對需要被復制到多個Reducer上，因此稱為Replicated Join．該算法犧牲部分網(wǎng)絡I?O代價來換取MRJ的初始化以及HDFS讀寫代價．到多個Reducer而導致較高的網(wǎng)絡I?O代價．文獻［5－6］將n元連接劃分為若干個組，每組涉及若干個關系并由一個作業(yè)完成，而后采用Replicated Join連接所有組生成的中間結果，然而該方法因為要窮舉所有可能的m（m＜n）元連接作為候選集而導致算法復雜度較高，從而使其應用范圍較窄．此外，上述所有研究確定的執(zhí)行計劃都只確定了連接的執(zhí)行順序，并未考慮無依賴關系的連接操作間的并行執(zhí)行策略．

本文首先基于MapReduce特性提出多元連接順序的確定方法，該方法復雜度較低且能夠很好地均衡中間結果的存儲代價與網(wǎng)絡傳輸代價．確定連接順序后，本文還給出一種算法來確定無依賴關系的連接操作間的并行執(zhí)行順序，該算法通過對節(jié)點資源的充分利用來提高多元連接的執(zhí)行效率．

對于第2類研究，文獻［12］針對theta－join提出一種隨機算法1－Bucket－Theta以及它的一個擴展算法M－Bucket；文獻［13］對文獻［12］中提出的算法進行下界分析，并通過聚類方法提高了M－Bucket算法的效率．文獻［14－15］總結現(xiàn)有實現(xiàn)算法為Map Join，Reduce Join，Semi Join等，這些算法分別適用于不同的查詢場景，如Map Join僅適用于數(shù)據(jù)量較小的關系能夠裝入內(nèi)存的查詢．MapReduce連接算法的優(yōu)化研究相對比較成熟，不在本文的研究范圍之內(nèi)．因此，不失一般性，本文采用沒有任何約束條件的通用的Reduce Join作為研究對象．

連接運算的執(zhí)行效率依賴于實現(xiàn)算法和運行環(huán)境，由此衍生出第3類研究．文獻［16］基于MapReduce提出一個改進的執(zhí)行框架Map－Reduce－Merge，新添加的Merge階段為Reduce Join的執(zhí)行節(jié)省了一次作業(yè)．文獻［17］對MapReduce框架進行修改，允許不同操作間的數(shù)據(jù)管道式傳輸，支持在線聚集以及持續(xù)查詢，然而改進后的框架使得失效恢復（fail recovery）機制變得非常復雜，且對于批處理性能的提高也很不明顯．文獻［7－9］給出了通用的MapReduce負載均衡方法，但他們都只考慮了任務的輸入代價，不適用于連接運算，因為它的輸出代價也不可忽略．本文提出的MapReduce負載均衡方法著重考慮連接任務的負載特性，與傳統(tǒng)的負載均衡方法有所不同．具體來講，本文提出的負載均衡方法基于的Reduce Join的Map階段僅負責將參與連接的數(shù)據(jù)表中的記錄解析成Key－Value對（此時I?O操作很少），并通過Shuffle階段傳輸?shù)綄腞educe任務中，而真正的連接操作是在Reduce階段中完成的（此時會產(chǎn)生大量I?O操作）．考慮到I?O代價是影響連接查詢的主要因素，我們對產(chǎn)生大量I?O操作的Reduce階段進行讀寫分析，綜合考慮Reduce任務的輸入和輸出代價及其對應的讀寫權重，最終基于這一綜合代價給出了Reduce任務的負載均衡方法．文獻［18］中實現(xiàn)的負載均衡是通過均衡數(shù)據(jù)塊實現(xiàn)的，本文與其有本質上的區(qū)別．

2 連接執(zhí)行計劃

多元連接執(zhí)行計劃的解空間很大，短時間內(nèi)很難找到最優(yōu)解．本節(jié)首先通過查詢樹模型確定解的一個子空間，而后通過復雜度較小的啟發(fā)式算法從中找到一個近似最優(yōu)解，并在此基礎上做進一步的優(yōu)化以均衡中間結果的存儲代價與網(wǎng)絡傳輸代價．最后，根據(jù)MapReduce框架特性給出一種算法來確定無依賴關系的連接操作間的并行執(zhí)行順序，該算法通過對節(jié)點資源的充分利用來提高多元連接的效率．

2．1 查詢樹模型

多元連接可以用一個查詢圖G＝（V，E）來表示［46］，其中V是節(jié)點的集合，每個節(jié)點代表一個關系（記為Ri），E是邊的集合，每條邊?Ri，Rj?連接2個之間存在連接屬性（A，B，C等）的關系，如圖1所示．圖1（a）所示的查詢圖包含6個關系，為6元連接；同理，圖1（b）所示的查詢圖為8元連接．

Fig．1 Example queries of multi－way join．圖1 多元連接查詢示例

多元連接執(zhí)行計劃的最優(yōu)解確定是個P完全問題［6］，傳統(tǒng)優(yōu)化方法通常采用查詢樹模型限定解的一個子空間，并設計算法從中獲取一個最優(yōu)解．如圖2所示，主流的查詢樹模型有Left－deep Tree，Right－deep Tree，Zigzag Tree，Bushy Tree［19］四種，其中前3種為順序執(zhí)行，最后1種為并行執(zhí)行．很多研究工作［4－5］顯示，并行執(zhí)行的Bushy Tree更適用于分布式環(huán)境．從圖2也可以看出，只有基于Bushy Tree確定的連接順序中不同連接操作間不是完全的依賴關系，是可以并行的．因此，本文選取Bushy Tree作為MapReduce多元連接的查詢樹模型．

Fig．2 Query trees．圖2 查詢樹模型

2．2 查詢樹模型

一個n元連接的執(zhí)行方式有2種：1）將其拆分為n－1個2元連接分別執(zhí)行，每個2元連接對應一個作業(yè)；2）在一個作業(yè)中執(zhí)行所有連接操作．然而，當n較大時，第1種執(zhí)行方式的作業(yè)初始化代價以及中間結果的存儲代價也隨之增大，第2種執(zhí)行方式也因為數(shù)據(jù)的多次傳輸而產(chǎn)生較大的網(wǎng)絡代價．為解決該問題，本文首先基于Bushy Tree初步確定多元連接的執(zhí)行順序，而后根據(jù)是否受益將部分2元連接合并成多元連接．

1）基于I?O代價的連接順序確定方法

通過Bushy Tree確定多元連接的執(zhí)行順序首先需要給出樹的構建規(guī)則．考慮到連接運算屬于I?O密集型計算，連接代價以I?O代價為主，本文針對MapReduce特性給出連接運算的I?O代價模型，并選擇I?O代價最小的連接順序．

正如第1節(jié)中的描述，本文選擇Reduce Join作為連接算法，下面以2元連接為例對其進行簡單介紹．Reduce Join由Map階段和Reduce階段組成．Map階段主要完成如下操作：①Map任務讀?。ㄍǔ楸镜刈x）參與連接的2個關系；②以連接屬性為鍵、記錄為值，按鍵排序后輸出鍵值對到本地磁盤；③將中間結果通過網(wǎng)絡傳輸給Reduce任務．Reduce階段主要完成如下操作：①接收來自Map任務的鍵值對并按鍵排序；②執(zhí)行連接操作，并將結果寫入分布式文件系統(tǒng)（HDFS）．

基于該分析，我們給出關系Ri和Rj進行Reduce Join的I?O代價計算方法，見式（1）：

其中，C1，C2，C3分別為本地、網(wǎng)絡和HDFS的I?O代價權重，三者均與系統(tǒng)硬件相關（其中C3還與HDFS的副本個數(shù)設置有關），其值均可事先通過文件讀寫實驗測出（在本文的實驗環(huán)境中，副本個數(shù)為3，通過實驗測得3個參數(shù)的值分別為C1＝3．67s?GB，C2＝8．93s?GB，C3＝13．37s?GB，三者之間的比值為1：2．4：3．6）；｜Ri｜代表關系的基數(shù)．另外，Map階段操作②首先需要溢出寫文件，而后讀取并排序輸出，因此共需3次讀寫操作；Reduce階段的操作①使用內(nèi)存和磁盤進行混合式排序，因此我們用參數(shù)λ表示該混洗比例，其值可通過經(jīng)驗設定．

通過Bushy Tree確定連接順序時，我們每次從查詢圖G＝（V，E）中選擇I?O代價最小的連接操作執(zhí)行，而后更新圖G及其對應的關系的特征參數(shù)，直到執(zhí)行完所有連接運算（見算法1）．算法1的復雜度為O（log（｜V｜｜E｜）），小于文獻［5－6］的復雜度O（log（｜V｜2｜E｜））．

算法1．PMC算法．?＊基于MC（minimal cost）的執(zhí)行計劃（plan）確定算法＊?

輸入：G＝（V，E），query profile；?＊包括關系的基數(shù)、連接屬性的基數(shù)等相關參數(shù)＊?

輸出：Bushy Tree．

PMC算法中計算最小代價時，｜Ri｜和｜Rj｜均已知，｜RiRj｜未知，需要我們計算．目前關于｜RiRj｜的計算方法通常假設Ri和Rj在連接屬性A上均勻分布［4，9］，具體計算方法見式（2）：

其中，｜A｜為連接屬性的基數(shù)．然而，事實上Ri和Rj通常不滿足均勻分布這一假設，因此在實際應用中，我們應該考慮數(shù)據(jù)傾斜因素．設Fi和Fj為A在Ri和Rj中出現(xiàn)的頻數(shù)分布，則定義傾斜度如下：

定義1．傾斜度．定義傾斜度δ為頻數(shù)分布Fi和Fj偏離均勻分布的程度，表達式見式（3）：

在實際計算中，傾斜度可以通過采樣獲?。辛藘A斜度，｜RiRj｜的計算方法見式（4）：

因為

考慮傾斜度因素計算出的｜RiRj｜更精確，同時基于最小代價的PMC算法確定的連接順序也更優(yōu)．

2）基于Replicated Join的優(yōu)化

通過Bushy Tree確定的執(zhí)行順序僅包含2元連接，這樣的執(zhí)行計劃會導致較高的作業(yè)初始化代價和中間結果的存儲代價．考慮到算法1在復雜度上的優(yōu)越性，我們保留由它確定的執(zhí)行順序，并在此基礎上做進一步的優(yōu)化．

解決上述問題的直觀想法為減少作業(yè)個數(shù)，也即增加每個作業(yè)執(zhí)行的連接操作個數(shù)．Replicated Join滿足該需求，但該算法中同一個鍵值對需要被復制到多個Reduce任務上，網(wǎng)絡代價較高．為此，本文分別計算查詢圖采用Replicated Join以及采用多個2元連接這2種執(zhí)行方法的I?O代價，定義“受益（benefit）”為二者的代價差．當受益為正時，合并這些2元連接，如圖3所示．2元連接的I?O代價見式（1），下面給出Replicated Join的I?O代價計算方法．

設查詢圖G＝（V，E），關系集合V＝｛R1，R2，…，Rn｝，E關聯(lián)的連接屬性集合為E－，Ri關聯(lián)的連接屬性集合為E－i，Replicated Join的I?O代價計算見式（5）：

Fig．3 Optimization based on replicated join．圖3 基于Replicated Join的優(yōu)化

基于Replicated Join的優(yōu)化需要對PMC算法確定的Bushy Tree進行遍歷，以合并所有可能的2元連接．然而，這樣做的代價很高，本文考慮到對無依賴關系的連接操作執(zhí)行Replicated Join明顯會導致較高的網(wǎng)絡I?O代價，因此僅判定具有依賴關系的連接操作（也即圖3（a）中只能順序執(zhí)行的子樹）．另外，如果一個順序執(zhí)行的子樹進行Replicated Join時受益為負，那么包含該子樹的順序執(zhí)行子樹的受益也為負．通過以上方法能夠大大降低樹的遍歷代價．基于Replicated Join的優(yōu)化算法見算法2．

OPTB的算法復雜度小于PMC算法確定的Bushy Tree中所有最大順序執(zhí)行子樹的高度之和．又考慮到順序執(zhí)行子樹的最大高度為n－1，故OPTB的算法復雜度為O（n）．

2．3 并行執(zhí)行順序

很多關于多元連接執(zhí)行計劃的優(yōu)化研究［4－6］都只確定了連接的執(zhí)行順序，并未考慮MapReduce環(huán)境下無依賴關系的連接操作間的并行執(zhí)行策略．

若圖3（b）為2．2節(jié)中最終優(yōu)化的多元連接順序，那么連接操作R2R7，R4R8，R3R5R6之間無依賴關系，可以并行執(zhí)行．下面結合MapReduce特性對并行執(zhí)行的優(yōu)勢進行分析．

MapReduce集群中每個節(jié)點最多可執(zhí)行的Map任務個數(shù)是預設的，因此最多可并行執(zhí)行的Map任務數(shù)也是確定的．對于連接運算，Reduce Join中的Map任務負責將關系中的元組按照連接屬性值進行分區(qū)，其執(zhí)行時間僅取決于處理數(shù)據(jù)量．又因為每個Map任務處理一個固定大小的分片，我們可以認為同時分配的Map任務同時結束．設MapReduce每次最多可并行的Map任務個數(shù)為M，M個任務的并行執(zhí)行稱為一輪［2，20－21］．若每輪Map任務的執(zhí)行時間為T，作業(yè)Ji需要的Map任務個數(shù)Mi＝ai×M＋bi，其中ai，bi∈NN，bi∈［0，M），那么可以認為Ji的執(zhí)行時間為（ai＋ bi?M ）T．若作業(yè)Jj需要的Map任務個數(shù)Mj＝aj×M＋bj，那么當bi＋bj≤M時，2個作業(yè)的并行執(zhí)行時間為（ai＋aj＋1）T，而串行執(zhí)行時間為（ai＋aj＋2）T，此時并行執(zhí)行可節(jié)省T時間；當bi＋bj＞M時，并行執(zhí)行時間和串行執(zhí)行時間同為（ai＋aj＋2）T．綜上，當bi＋bj≤M時，作業(yè)Ji和Jj并行執(zhí)行的效率高于串行．

當有多個作業(yè)滿足這一條件時，我們選取bi＋bj最大的2個作業(yè)并行執(zhí)行，從而充分利用計算資源．下面給出具體的算法實現(xiàn)：

2．4 小 結

通過2．2節(jié)和2．3節(jié)給出的優(yōu)化算法，我們最終動態(tài)確定多元連接的執(zhí)行計劃．圖4以流程圖的方式描述了執(zhí)行計劃的優(yōu)化步驟．

Fig．4 Optimization flow of the execution plan．圖4 執(zhí)行計劃優(yōu)化流圖

給定查詢圖G，首先通過PMC算法初步確定Bushy Tree，而后分別通過算法OPTB和PEE進行優(yōu)化，直到更新后的樹中葉子節(jié)點的個數(shù)小于3．

3 負載均衡

良好的執(zhí)行計劃固然重要，但對執(zhí)行框架的優(yōu)化同樣可以有效地提高多元連接的執(zhí)行效率，這一點在分布式環(huán)境中尤為重要．一種公平的并行任務負載分配方法可以有效地減少MapReduce中的“短板效應”，從而提高連接操作內(nèi)部的并行度．

連接運算的MapReduce實現(xiàn)算法有很多，分別適用于不同的查詢場景．不失一般性，本文選擇沒有任何約束條件的Reduce Join作為連接執(zhí)行算法．該算法包括Map和Reduce 2個階段，Map階段只負責將關系中的元組按照連接屬性值進行分區(qū)以輸出到不同的Reduce任務，運算完全相同，因此Map任務的負載完全取決于處理數(shù)據(jù)量．又因為MapReduce中每個Map任務只負責處理一個數(shù)據(jù)分片（split，默認64MB），所以Map階段各個Map任務是負載均衡的．很多研究工作也都作出Map任務均衡的假設，如文獻［6，22］．因此，本文僅研究連接運算的Reduce任務負載均衡．

Reduce Join算法在Reduce階段執(zhí)行連接運算，連接屬性值的不均勻分布將會導致由默認Hash分區(qū)函數(shù)確定的Reduce任務負載不均衡．為提高Reduce任務間的并行度，本節(jié)給出一種針對連接運算的負載均衡優(yōu)化方法．

3．1 負載均衡模型

設R1和R2為參與連接的2個關系，連接屬性值的集合記為A，A在R1和R2中的頻數(shù)分布分別為F1和F2．在計算Reduce任務的負載之前，我們先給出連接屬性值a∈A的負載貢獻定義如下：

定義2．負載貢獻．連接屬性值a∈A的負載貢獻（LCa）是指執(zhí)行該連接操作的代價，見式（6）：

LCa＝ω1（f1a＋f2a）＋ω2（f1a×f2a），（6）其中，f1a和f2a分別為R1和R2中連接屬性值為a的元組個數(shù)；ω1和ω2為Reduce任務輸入數(shù)據(jù)和輸出數(shù)據(jù)的處理代價權重，輸入數(shù)據(jù)為網(wǎng)絡I?O，輸出數(shù)據(jù)寫到HDFS上，二者的比值是由運行多元連接的分布式集群系統(tǒng)決定的．此處，我們認為連接運算代價中I?O占主導地位，CPU處理代價可以忽略，文獻［5－6］中也有同樣結論．文獻［8］給出的當前最好的負載均衡方法中采用的代價模型僅考慮輸入數(shù)據(jù)對Reduce任務負載的影響，而事實上對于連接運算輸出數(shù)據(jù)的代價不容忽略．

通過對MapReduce運行機制的分析可知，Reduce任務的負載取決于分區(qū)函數(shù)．分區(qū)函數(shù)將連接屬性值劃分成若干個組，每組對應一個Reduce任務．設分區(qū)函數(shù)將連接屬性值的集合A劃分為A1，A2，…，AR一共R個組，那么組Ai的處理代價

第i個Reduce任務的負載Load（Ri）＝Load（Ai），Reduce任務負載均衡這一目標可以等價表示如下：

負載均衡模型中最關鍵的是獲取連接屬性A在R1和R2中的頻數(shù)分布F1和F2．獲取這2個分布，最精確的方法是對不同鍵值進行頻數(shù)統(tǒng)計［23］，但當鍵值個數(shù)很多時，會耗費大量存儲，且在匯總各個Map任務的統(tǒng)計信息時還會帶來很高的網(wǎng)絡傳輸代價．針對該問題，文獻［5，7－9］對鍵值進行Hash從而降低統(tǒng)計信息的規(guī)模．然而，文獻［9］在Map任務執(zhí)行的同時對頻數(shù)信息進行統(tǒng)計，這樣會導致第2輪Map任務無法執(zhí)行，還會造成數(shù)據(jù)到Reduce的傳輸延遲，因為必須等到根據(jù)頻數(shù)信息確定Partition函數(shù)后才能進行傳輸．文獻［5，7－8］則單獨開啟一個作業(yè)進行頻數(shù)統(tǒng)計，避免了上述問題．基于以上分析，本文可以采用類似的方法獲取連接屬性值的頻數(shù)信息，并據(jù)此確定Reduce任務的個數(shù)以及Partition函數(shù)．

3．2 負載均衡算法

文獻［6］指出Reduce任務的負載均衡是一個NP難問題，不能夠在多項式時間內(nèi)獲取最優(yōu)解，因此我們僅專注于尋找盡可能接近最優(yōu)解的近似解．由于連接屬性值的不可分割性，擁有相同連接屬性值的鍵值對必須發(fā)送到同一Reduce任務節(jié)點進行連接運算，Reduce任務的負載均衡問題可以轉換成盡可能降低Reduce任務的最大負載max｛Load（Ri）｝．

理想情況下，所有Reduce任務的負載完全相同，此時max｛Load（Ri）｝＝Avg｛Load（Ri）｝．然而，這種情況不總發(fā)生，本文給出一種樸素的均衡算法來獲取近似解．該算法首先對A中所有連接屬性值的負載貢獻值按降序排序，然后每次將連接屬性值為a的鍵值對分配給當前負載最小的Reduce任務（詳見算法4）．

為評估算法4，我們將由該算法獲取的Reduce任務負載最大值Lmax與最優(yōu)算法獲取的最大值L＊max進行對比，并得出Lmax的上界如下：Lmax≤1．5L＊max．

本文給出的負載均衡方法是以等值連接為例進行描述的，它還適用于其他連接，也可以擴展到連接以外的其他類型作業(yè)．例如，進行近似連接時，只需將連接屬性值a（鍵值對中的鍵）替換為一個滿足近似條件（如｜a1－a2｜≤δ）的2元組?a1，a2?，并將負載貢獻值的計算公式中f1a和f2a分別替換為R1中連接屬性值為a1的元組個數(shù)以及R2中連接屬性值為a2的元組個數(shù)．執(zhí)行Replicated Join時，鍵值對中的鍵將會變成多個連接屬性構成的多元組．對于連接以外的其他作業(yè)，我們可以將Reduce任務輸入數(shù)據(jù)的處理代價函數(shù)（頻數(shù)的加和）以及輸出數(shù)據(jù)的處理代價函數(shù)（頻數(shù)的乘積）進行適應性的更改．

3．3 Reduce任務個數(shù)的確定

現(xiàn)有通用的Reduce端負載均衡的方法［7－9］均未考慮Reduce任務個數(shù)的確定方法，本文根據(jù)獲取的鍵值頻數(shù)統(tǒng)計信息給出一種簡單的確定規(guī)則．

設Map任務的輸出中不同鍵值的個數(shù)為k，所有鍵值的負載貢獻和為Sum，其中鍵值的最大負載貢獻為LCmax，Reduce任務個數(shù)為R，通過優(yōu)化算法獲取的Reduce任務最大負載為Lmax．當LCmax≥Sum?R時，也即R≥Sum?LCmax時，Lmax的取值不再下降，始終為LCmax，這意味著作業(yè)的性能不再提高，而它的資源消耗卻隨著R的增加而增大．因此，有必要找到R的一個臨界值使得連接的執(zhí)行效率最高．另外，考慮到Lmax的取值還與Sum有關，本文給出均衡算法的度量函數(shù)g（Lmax，R）的表達式如下：

其中，α是性能與能耗之間的權重比，度量值越小，均衡效果越好．函數(shù)g（Lmax，R）應該存在一個極小值點R0，使得在該點處性能與資源消耗達到一個很好的折中，且該值有可能比Sum?LCmax小．4．2節(jié)中通過大量實驗得出，當α＝0．05時，函數(shù)g（Lmax，R）的極小值點正好就是Lmax不再下降的臨界值．

另外，考慮到不同鍵值的個數(shù)可能會很大，這將導致頻數(shù)統(tǒng)計信息不能存入內(nèi)存，針對該問題，我們可以采用文獻［8］中提出的optimal sketch packing算法，該方法通過Hash函數(shù)將鍵值（這里是負載貢獻值）進行哈希后再進行均衡分配，從而降低鍵的規(guī)模，節(jié)省統(tǒng)計信息占用的內(nèi)存．本質上，該方法是犧牲精確度來降低統(tǒng)計信息的存儲空間．

4 實驗與范例分析

本節(jié)設計實驗對提出的連接執(zhí)行計劃優(yōu)化方法以及連接負載均衡方法進行驗證和分析．其中，4．1節(jié)中的實驗是依托圖1中給出的查詢圖生成的虛擬數(shù)據(jù)表進行多元連接查詢設計的，該實驗能夠很好地驗證本文提出的執(zhí)行計劃的優(yōu)化效果；4．2節(jié)中的實驗則是依托TPC－H數(shù)據(jù)集中提供的邏輯數(shù)據(jù)表進行設計的，本文通過控制其數(shù)據(jù)的生成方式來設計實驗以驗證文中提出的連接負載均衡方法．實驗的具體設置在4．1節(jié)和4．2節(jié)中均有對應的詳細描述．

4．1 執(zhí)行計劃的優(yōu)化效果分析

以圖1中查詢圖為例對本文提出的執(zhí)行計劃優(yōu)化方法進行效果分析，相應的特征參數(shù)見表1和表2．

Table 1 Cardinalities of the Relations in Fig．1（a）表1 圖1（a）對應的關系特征參數(shù)

Table 2 Cardinalities of the Relations in Fig．1（b）表2 圖1（b）對應的關系特征參數(shù)

首先，為了驗證本文提出的連接順序確定PMC算法的優(yōu)化效果，本文將其與最優(yōu)解（可用分支限定法獲取）進行對比．圖5中，POPT代表最優(yōu)解，從圖5可以看出PMC的代價比最優(yōu)解稍高，二者的比值分別為1．003和1．034．由此可見，PMC算法能夠確定一個很好的連接順序，從而降低連接的I?O代價．

接著，我們分析了2種查詢圖下算法OPTB和PEE的優(yōu)化效果．從表3可以看出，OPTB算法能夠找出可以合并為多元連接的2元連接，一定程度上降低了I?O代價；PEE算法能夠找出最大限度使用集群計算資源的可并行連接操作，從而節(jié)省多元連接的整體運行時間．

Fig．5 I?O cost comparison of PMCand POPT．圖5 PMC算法與POPT的I?O代價對比

Table 3 Optimization Results of Algorithms OPTBand PEE表3 算法OPTB和PEE的優(yōu)化效果

4．2 負載均衡方法驗證

文獻［5，7－9］中均提到采用模擬實驗驗證其提出的負載均衡方法，假設鍵值服從Zipf分布．不失一般性，本文也采用該方法來驗證第3節(jié)中針對連接運算設計的負載均衡方法．以等值2元連接為例，假設參與連接的2個關系表R1和R2服從相同的Zipf分布，數(shù)據(jù)條數(shù)分別為｜R1｜和｜R2｜，不同連接屬性值的個數(shù)為k，則R1和R2中第i個最頻繁出現(xiàn)的連接屬性值的出現(xiàn)概率pi＝1?（iz×Hk），其中z代表數(shù)據(jù)的傾斜度，Hk是調(diào)和系數(shù)，那么第i個連接屬性值的負載貢獻值計算如下：

依托TPC－H數(shù)據(jù)集中提供的邏輯數(shù)據(jù)表，我們設計具體的測試用例為：｜R1｜＝109，｜R2｜＝2×109，k＝3×104，3×105，3×106，z的取值范圍為［0，1］．這里需要指出的是，z僅代表關系R1和R2中連接屬性值的傾斜程度，并不代表負載貢獻值集合的傾斜度（記為δ），但δ與z值是成正相關的，且比z大．

首先，我們對比不同因素下負載均衡算法的效果，采用IR（imbalance ratio）來度量，它是所有Reduce任務中的最大負載（Lmax）與平均負載（Avg＝Sum?R）之間的比值．從圖6可以看出，IR的影響因素有δ（受z影響），k和R，與δ和R成正相關，與k成負相關．從圖6我們還可以看出，IR的值早在z＝0．5，R＝64時已經(jīng)超過1．5，這是因為IR是Lmax與Avg之間的比值，而通過最優(yōu)負載均衡算法獲取的L＊max通常會因為δ較大而遠大于Avg．例如，當一個鍵的頻數(shù)占總頻數(shù)的80%時，因為鍵的不可分割性，Lmax和L＊max會遠大于Avg．

其次，為了驗證本文設計的負載均衡算法的有效性，我們將其得到的最大負載值與默認Hash函數(shù)得到的最大負載值進行對比．從圖7可以看出，在3種傾斜度、3種Reduce個數(shù)下，我們的算法均比默認Hash的好．從圖7（a）可以看出，隨著Reduce個數(shù)的增加，負載均衡算法的優(yōu)勢越來越明顯；而在圖7（b）和圖7（c）中，Reduce個數(shù)為100和1 000時，均衡算法得到的最大負載值均未發(fā)生變化，這是因為此時數(shù)據(jù)太過傾斜而產(chǎn)生了“二八現(xiàn)象”，圖7（b）和圖7（c）對應的最大負載值分別在Reduce個數(shù)大于95以及16后不再發(fā)生變化（從圖8中可以看出），這也驗證了我們在3．3節(jié)中的理論分析．另外，在圖7（b）和圖7（c）中，雖然默認Hash得到的最大負載值依然隨著Reduce個數(shù)的增加而降低，但該值由于數(shù)據(jù)傾斜以及鍵值不可分割等原因而不會低于均衡算法得到的值．

最后，為了驗證3．3節(jié)中提出的Reduce任務個數(shù)的確定方法，我們分析了正規(guī)化后的最大負載Lmax與負載均衡算法的評估函數(shù)g（Lmax，R）隨Reduce任務個數(shù)R的變化情況．通過大量實驗，我們發(fā)現(xiàn)當函數(shù)g（Lmax，R）中的α＝0．05時，它的極小值點正好就是Lmax不再下降的臨界值R0．另外，在實驗過程中，我們發(fā)現(xiàn)z值越大，臨界值R0的下降趨勢越不明顯，因此，為直觀起見，本文選取了其中5個具有代表性的z值進行展示．從圖8（a）可以看出，隨著R的增長，Lmax不斷減小，最終趨向平穩(wěn)值，圖中5個z值對應的臨界R值分別為974，95，16，6，4，這與我們通過均衡算法度量函數(shù)g（Lmax，R）得到的極小值點是完全吻合的（見圖8（b））．

Fig．6 Imbalance ratios of the load balancing algorithm under different skew degrees．圖6 不同傾斜度下負載均衡算法的IR比較

Fig．7 Max load comparison between the load balancing algorithm and the default Hash under 3million keys．圖7 k＝3×106時不同傾斜度下負載均衡算法與默認Hash的最大負載對比

Fig．8 Relationships between the formalized Lmax，g（Lmax，R）and Runder 3million keys．圖8 k＝3×106時正規(guī)化的最大負載Lmax以及函數(shù)g（Lmax，R）隨R的變化曲線

5 總 結

本文基于MapReduce研究多元連接的優(yōu)化方法，主要從以下2部分展開研究：連接的執(zhí)行計劃和連接的負載均衡．

針對前者，本文首先分析現(xiàn)有主流的查詢樹模型，確定適合本文研究環(huán)境的Bushy Tree；隨后通過白盒分析給出MapReduce連接算法的I?O代價模型，并選擇I?O代價最小的連接順序作為初步的執(zhí)行計劃；接著對執(zhí)行計劃做進一步的優(yōu)化，根據(jù)是否受益將查詢樹中的2元連接合并成Replicated Join，以降低多個作業(yè)引起的中間結果代價；最后結合MapReduce特性提出一種作業(yè)并行執(zhí)行算法，以提高集群資源的使用率．

針對后者，本文首先分析連接運算的特性，給出連接負載的定義以及負載均衡目標；接著給出具體的均衡算法，并證明該算法的上界；最后在實驗中分析Reduce任務個數(shù)的確定與性能之間的關系．

實驗證明，本文提出的連接執(zhí)行計劃以及負載均衡的優(yōu)化算法是有效的．本研究對大數(shù)據(jù)環(huán)境下MapReduce多元連接的應用具有指導意義，可以優(yōu)化如OLAP分析中的星型連接，社交網(wǎng)絡中社團發(fā)現(xiàn)的鏈式連接等應用的性能．

［1］Han Xixian，Yang Donghua，Li Jianzhong．Approximate join aggregate on massive data［J］．Chinese Journal of Computers，2010，33（10）：1919 1933（in Chinese）（韓希先，楊東華，李建中．海量數(shù)據(jù)上的近似連接聚集操作［J］．計算機學報，2010，33（10）：1919 1933）

［2］Dean J，Ghemawat S．MapReduce：Simplified data processing on large clusters［J］．Communications of the ACM，2008，51（1）：107 113

［3］Doulkeridis C，Norvag K．A survey of large－scale analytical query processing in MapReduce［J］．The VLDB Journal，2013，23（3）：355 380

［4］Chen M S，Yu P S，Wu K．Optimization of parallel execution for multi－join queries［J］．IEEE Trans on Knowledge and Data Engineering，1996，8（3）：416 428

［5］Wu S，Li F，Mehrotra S，et al．Query optimization for massively parallel data processing［C］??Proc of the 2nd ACM Symp on Cloud Computing．New York：ACM，2011：1 13

［6］Zhang X F，Chen L，Wang M．Efficient multi－way theta－join processing using MapReduce［J］．Proceedings of the VLDB Endowment，2012，5（11）：1184 1195

［7］Gufler B，Augsten N，Reiser A，et al．Load balancing in MapReduce based on scalable cardinality estimates［C］??Proc of the Int Conf on Data Engineering．Piscataway，NJ：IEEE，2012：522 533

［8］Yan W，Xue Y，Malin B．Scalable and robust key group size estimation for reducer load balancing in MapReduce［C］?? Proc of the IEEE Int Conf on Big Data．Piscataway，NJ：IEEE，2013：156 162

［9］Gufler B，Augsten N，Reiser A，et al．Handling data skew in MapReduce［C］??Proc of the 1st Int Conf on Cloud Computing and Services Science．Boca Raton，F(xiàn)lorida：CRC Press，2011：574 583

［10］Zhou M Q，Zhang R，Zeng D D，et al．Join optimization in the MapReduce environment for column－wise data store［C］??Proc of the 6th Int Conf on Semantics Knowledge and Grid．Piscataway，NJ：IEEE，2010：97 104

［11］Afrati F N，Ullman J D．Optimizing multiway joins in a Map－Reduce environment［J］．IEEE Trans on Knowledge and Data Engineering，2011，23（9）：1282 1298

［12］Okcan A，Riedewald M．Processing theta－joins using MapReduce［C］??Proc of the ACM SIGMOD Int Conf on Management of Data Athens．New York：ACM，2011：949 960

［13］Koumarelas I K，Naskos A，Gounaris A．Binary theta－joins using MapReduce：Efficiency analysis and improvements［C］??Proc of the Workshops of the EDBT ICDT 2014Joint Conf．Boca Raton，F(xiàn)lorida：CRC Press，2014：6 9

［14］Blanas S，Patel J M，Ercegovac V，et al．A comparison of join algorithms for log processing in MapReduce［C］??Proc of the ACM SIGMOD Int Conf on Management of Data．New York：ACM，2010：975 986

［15］Luo G，Dong L．Adaptive join plan generation in Hadoop，NC27705［R］．Durham，NC：Duke University，2010

［16］Yang H，Dasdan A，Hsiao R，et al．Map－reduce－merge：Simplified relational data processing on large clusters［C］?? Proc of the ACM SIGMOD Int Conf on Management of Data．New York：ACM，2007：1029 1040

［17］Condie T，Conway N，Alvaro P，et al．Online aggregation and continuous query support in MapReduce［C］??Proc of the ACM SIGMOD Int Conf on Management of Data．New York：ACM，2010：1115 1118

［18］Ding Youwei，Qin Xiaolin，Liu Liang，et al．An energy efficient algorithm for big data processing in heterogeneous cluster［J］．Journal of Computer Research and Development，2015，52（2）：377 390（in Chinese）（丁有偉，秦小麟，劉亮，等．一種異構集群中能量高效的大數(shù)據(jù)處理算法［J］．計算機研究與發(fā)展，2015，52（2）：377 390）

［19］Aljanaby A，Abuelrub E，Odeh M．A survey of distributed query optimization［J］．Int Arab Journal of Information Technology，2005，2（1）：48 57

［20］Agrawal P，Kifer D，Olston C．Scheduling shared scans of large data files［J］．Proceedings of the VLDB Endowment，2008，1（1）：958 969

［21］Li F，Ooi B C，Ozsu M T，et al．Distributed data management using MapReduce［J］．ACM Computing Surveys，2014，46（3）：31：1 42

［22］Nykiel T，Potamias M，Mishra C，et al．MRShare：Sharing across multiple queries in MapReduce［J］．Proceedings of the VLDB Endowment，2010，3（1）：494 505

［23］Ibrahim S，Jin H，Lu L，et al．LEEN：Locality?fairnessaware key partitioning for MapReduce in the cloud［C］??Proc of the 2nd IEEE Int Conf on Cloud Computing Technology and Science．Piscataway，NJ：IEEE，2010：17 24 Li Tiantian，born in 1989．PhD candidate．Student member of China Computer Federation．Her main research interests include energy efficient computing，and data intensive computing．

Yu Ge，born in 1962．Professor and PhD supervisor in Northeastern University．His main research interests include database theory and data flow．

Guo Chaopeng，born in 1990．Master．His main research interests include iterative computing，and data intensive computing．

Song Jie，born in 1980．PhD and associate professor in Northeastern University．His main research interests include cloud computing，data intensive computing and big data．

Multi－Way Join Optimization Approach Based on MapReduce

Li Tiantian1，Yu Ge1，Guo Chaopeng2，and Song Jie21（College of Computer Science and Engineering，Northeastern University，Shenyang110819）2（Software College，Northeastern University，Shenyang110819）

Multi－way join is one of the most common data analysis operations，and MapReduce programming model that has been widely used to process large scale data sets has brought new challenges to multi－way join optimization．Traditional optimization approaches cannot be simply adapted to fit MapReduce feature，so there is still optimization room for MapReduce join algorithm．As to the former，we think I?O is the main cost of join．This paper first proposes an I?O cost based heuristic algorithm to initially determine a join sequence，and conducts further optimization．After the optimization，we also design a parallel execution algorithm to improve the whole performance of multiway join．As to the latter，we think load balancing can effectively decrease the“buckets effect”of MapReduce．This paper proposes a fair task load allocation algorithm to improve the intra－join parallelism，and also analyzes the method to decide the appropriate number of Reduce tasks．Experiments verify the effectiveness of the proposed optimization approaches．This study contributes to multi－way join applications in big data environment，such as the star－join in OLAP and the chainjoin in social network．

multi－way join；execution plan；I?O cost；performance optimization；MapReduce programming model；load balancing

TP393

2014－11－24；

2015－03－04

國家自然科學基金重大項目（61433008）；國家自然科學基金青年基金項目（61202088）；國家博士后科學基金面上項目（2013M540232）；中央高?；究蒲袠I(yè)務費專項基金項目（N120817001）；教育部高等學校博士學科點博導基金項目（20120042110028）

This work was supported by the Major Program of the National Natural Science Foundation of China（61433008），the National Natural Science Foundation for Young Scholars（61202088），the Science Foundation of China for Post－doctor（2013M540232），the Fundamental Research Funds for the Central Universities（N120817001），and the PhD Programs Foundation of Ministry of Education of China（20120042110028）．