MapReduce中Combine優化機制的利用

賈歐陽 阮樹驊 田興 楊峻興 李丹

摘 要： 由Apache軟件基金會開發的Hadoop分布式系統基礎架構，作為一個主流的云計算平臺，其核心框架之一的MapReduce性能已經成為一個研究熱點，其中對于Shuffle階段的優化，使用Combine優化機制是關鍵。文章詳細介紹了MapReduce計算框架及Shuffle流程；分別從機理簡介、執行時機、運行條件三方面詳細闡述了如何利用Combine優化機制；通過搭建Hadoop集群，運用MapReduce分布式算法測試實驗數據。實驗結果充分證明，正確地運用Combine優化機制能顯著提高MapReduce框架的性能。

關鍵詞： 云計算； Hadoop； MapReduce； Shuffle； Combine

中圖分類號：TP393.2 文獻標志碼：A 文章編號：1006-8228（2013）09-01-04

0 引言

我們正處在一個數據爆炸的時代，數據生成速度之快令人驚訝：紐約證券交易所每天產生1TB的數據，Facebook存儲著約100億張照片占用了約1PB存儲容量[1]。由Apache軟件基金會開發的Hadoop分布式系統基礎架構，正是為了解決海量數據的存儲和計算問題，并且由于其高可靠性、高可擴展性、高效性和高容錯性，已經成為一種主流的云計算平臺[2]。Hadoop核心框架之一的MapReduce，在性能優化和提高等方面的問題已經被學術界和工業界所關注，而其中很重要的一部分是對Shuffle階段的優化。本文詳細介紹了MapReduce框架和Shuffle階段流程，研究分析了Shuffle優化過程中利用Combine優化機制存在的問題，通過實驗和理論分析找出了解決方案，提出了Combine優化機制的執行時機和運行條件，并利用實例數據充分證明了正確地利用Combine優化機制能顯著提高MapReduce框架性能。

1 MapReduce框架

1.1 框架簡介

MapReduce是一種能夠在普通配置計算機上并行處理大量數據的并行計算框架，使用這個框架，我們只需要編寫我們想要執行的簡單運算即可，寫出的程序自動在集群上并行執行，而不必關心集群之間的調度、數據分布、負載均衡、容災備份、通信等復雜細節，這些問題都已經被MapReduce框架封裝好了。

MapReduce可以讓沒有任何分布式、并行編程經驗的程序員很容易地利用分布式系統的資源[3]，用戶只需要自己定義map函數和reduce函數，這兩個函數的靈感來自Lisp和許多其他函數式語言的map和reduce原語[4]，map函數從輸入數據中獲取鍵/值對同時生成一個中間鍵值對集合，reduce函數合并有相同中間鍵的中間值，從而得到一個想要的結果。MapReduce框架會在任務運行時關注調度任務，并監視任務的執行狀況，如果執行失敗，將重新執行該任務[5]。

1.2 實現機制

首先輸入文件會被用戶程序調用的MapReduce庫切分成若干64MB大小的數據塊，這些數據塊在框架中會被自動地創建副本，然后調用fork原語創建數個子程序，在這些子程序中有一個特殊程序master，其他程序都被當作worker程序，一個空閑的worker程序將收到由master分配的一個map任務和reduce任務。

worker程序從輸入的數據片段中解析出鍵值對，然后這些鍵值對被傳遞給用戶自定義實現的map函數，由map函數處理后，生成的中間鍵值對最終被寫到本地磁盤上，master獲取到這些鍵值對在本地磁盤上的存儲位置并傳送給負責執行reduce任務的worker節點。

reduce worker程序使用RPC從負責執行map任務的worker節點所在主機的磁盤上讀取這些緩存數據，由于存在不同的key值映射到相同的reduce任務上，因此會進行排序，如果中間數據太大無法在內存中完成排序，那么就要進行外排序[5]，最終使得具有相同key值的數據聚合在一起。

排序后的中間數據經過reduce worker遍歷后，reduce worker程序將每一個惟一的中間key值和它相關的中間value值的集合傳遞給用戶自定義的reduce函數[5]。

當所有的map、reduce任務完成后，用戶程序對map、reduce的調用才返回。

2 Shuffle過程

Shuffle過程可以理解為是從map映射輸出到reduce消化輸入的整個過程，整個過程被稱為MapReduce的“心臟”，關乎整個框架性能[1]，應用開發人員對MapReduce框架的改進也主要集中在Shuffle階段。

MapReduce框架中Shuffle流程如圖1所示[1]。

2.1 Map端

⑴ 作業提交以后，Map端從輸入塊中讀取record，依次調用map函數進行處理，并映射輸出（key，value）鍵值對。

⑵ 生成的鍵值對串行化的輸出到環形內存緩沖區[7]，這期間如果應用程序員沒有定制partitioner，那么系統會調用默認的HashPartitioner把鍵值對劃分到對應的partition分區。

⑶ 當緩沖區的內容達到設定的閾值時，一個后臺spill線程便開始把這些內容溢寫到磁盤[1]，spill線程在把緩沖區數據寫到磁盤前，會對它進行一個二次快速排序，首先根據所屬的partition排序，然后在每個partition內再按key排序[8]。

⑷ 當緩沖區的數據輸出到磁盤后，可能會出現多個spill文件，這時候就進入Map端的合并（merge）階段，會對這些文件做一個歸并排序，最后輸出一個index文件和一個數據文件，index文件記錄了不同的key在數據文件中偏移量[7]。

2.2 Reduce端

⑴ Copy Phase：在Map端，只要有一個任務完成，Reduce任務就開始通過http方式復制其輸出，而不是等所有Map端任務都結束之后再進行[1]。

⑵ Sort Phase：Map端輸出在被取回的同時，被合并成多個文件，并按鍵排序，這個階段更恰當地說，應該只進行了合并，因為排序（Sort）是在Map端進行的[1]。

⑶ Merge Phase：歸并過程中的數據可能既有在內存中的也有在磁盤上的，如果都在內存中，則會直接被復制到Reduce端而省去向磁盤寫這一步[8]，然后通過歸并得到最終的文件。

3 Combine優化

3.1 Combine優化機制簡介

MapReduce框架的運作基于鍵值對，即數據的輸入是鍵值對，生成的結果也是存放在集合里的鍵值對，其中鍵值對的值也是一個集合，一個MapReduce任務的執行過程以及數據輸入輸出的類型如下所示，這里我們定義list表示集合：

map（K1， V1） -> list（K2， V2）

combine（K2， list（V2）） -> list（K2， V2）

reduce（K2， list（V2）） -> list（K3， V3）

map函數操作所產生的鍵值對會作為combine函數的輸入，經combine函數處理后再送到reduce函數進行處理，減少了寫入磁盤的數據量，同時也減少了網絡中鍵值對的傳輸量。在Map端，用戶自定義實現的Combine優化機制類Combiner在執行Map端任務的節點本身運行，相當于對map函數的輸出做了一次reduce[8]。

集群上的可用帶寬往往是有限的，產生的中間臨時數據量很大時就會出現性能瓶頸，因此盡量避免Map端任務和Reduce端任務之間大量的數據傳輸是很重要的。使用Combine機制的意義就在于使Map端輸出更緊湊，使得寫到本地磁盤和傳給Reduce端的數據更少[1]。

選用Combine機制下的Combiner雖然減少了IO，但是等于多做了一次reduce，所以應該查看作業日志來判斷combine函數的輸出記錄數是否明顯少于輸入記錄的數量，以確定這種減少和花費額外的時間來運行Combiner相比是否值得[8]。

在具體的分布式應用中，應用開發人員需要自己定制map函數和reduce函數，在本文實驗中用到的具體算法如下：

Algorithm 1是分割映射鍵值對的map算法。map函數的輸入分別為文本數據中行偏移量key、每行文本內容text和上下文對象context，算法第1行將輸入的每行文檔內容數據text分割成單個的單詞k，放入集合K中，第2-3行對K中的每個k，映射輸出鍵值對（k，1），這里每個鍵k對應的值是1。

Algorithm 2是對中間鍵值對合并的combine算法。combine函數的輸入為經map函數分割后的每個單詞k、單詞k及其統計次數存放的集合K1和上下文對象context，其中鍵值對的輸入鍵midsum為經過combine算法后分發出的一類鍵名，用以標識鍵值對已經經過combine算法處理。算法第1行首先判斷輸入鍵是否為midsum，如果不是，則第2-4行對K1集合中的統計次數進行累加，輸出鍵值對midsum作為鍵，累加結果sum作為值；第5-7行因為輸入鍵是midsum，則直接映射輸出，避免輪詢造成更大的開銷。

Algorithm 3是規約reduce算法。reduce函數的輸入可能是combine函數的輸出（midsum，sum）（存放于集合K2中），也可能是沒有經過combine函數合并的map函數的輸出（k，1） （仍然存放于集合K1中），還有上下文對象context，算法第1-2行對K1/K2集合中每個k進行累加求和，第3行輸出最終的鍵值對（endsum，sum），sum即為最終的求和結果。

算法1至3如圖2所示。

[Algorithm 1：分割鍵值對map算法

輸入：key， text， context

輸出：中間鍵值對（k，v）

步驟：

1 list（K） ← split text

2 foreach k ∈ K

3 emit intermediate（k，1）；][Algorithm 3：規約reduce算法

輸入：key， K1/K2， context

輸出：求和結果

步驟：

1 foreach k ∈ K1/K2

2 compute the sum of v

3 emit （endsum，sum）；＼&][Algorithm 2：合并中間鍵值對combine算法

輸入：key， K1， context

輸出：中間鍵值對（k，v）

步驟：

1 if k does not start with midsum

2 foreach k ∈ K1

3 compute the sum of v

4 emit intermediate（midsum，sum）；

5 else

6 foreach k ∈ K1

7 emit intermediate（k，v）；

3.2 Combine優化機制執行時機

⑴ Map端spill的時候

在Map端內存緩沖區進行溢寫的時候，數據會被劃分成相應分區，后臺線程在每個partition內按鍵進行內排序。這時如果指定了Combiner，并且溢寫次數最少為3（min.num.spills.for.combine屬性的取值）時，Combiner就會在排序后輸出文件寫到磁盤之前運行[1]。

⑵ Map端merge的時候

在Map端寫磁盤完畢前，這些中間的輸出文件會合并成一個已分區且已排序的輸出文件，按partition循環處理所有文件，合并會分多次，這個過程也會伴隨著Combiner的運行[7]。

⑶ Reduce端merge的時候

從Map端復制過來數據后，Reduce端在進行merge合并數據時也會調用Combiner來壓縮數據。Combiner通常被看作是一個Map端的本地reduce函數的實現類Reducer，這個實驗[9]也驗證了這一理論的不足，但是在很多情況下Combiner在Reduce端的作用是有限的。

3.3 Combine優化機制運行條件

⑴ 滿足交換和結合律[10]

結合律：

（1）+（2+3）+（4+5+6）==（1+2）+（3+4）+（5）+（6）== ...

交換律：

1+2+3+4+5+6==2+4+6+1+2+3== ...

應用程序在滿足如上的交換律和結合律的情況下，combine函數的執行才是正確的，因為求平均值問題是不滿足結合律和交換律的，所以這類問題不能運用Combine優化機制來求解。

例如：mean（10，20，30，40，50）=30

但mean（mean（10，20），mean（30，40，50））=22.5

這時在求平均氣溫等類似問題的應用程序中使用Combine優化機制就會出錯。

另外需注意，在撰寫含有Combiner的應用時，對于所有map函數的輸出，并非一定都經過Combiner，有些會直接進入Reducer。

如果我們在程序中定制了一個Combiner，MapReduce框架使用它的次數可能是0次也可能是多次，為了保證Combine機制的正確運用，Combiner在數據的轉換上必須與Reducer等價，如果我們去掉Combiner，Reducer的輸出也應保持不變，而且，當Combiner被應用于中間數據的任意子集時，仍需保持等價的轉換特性[9]。

⑵ Combine優化機制中存在的輪詢問題

在開發過程中，使用Combine優化機制會存在輪詢問題，即一個combine函數的輸出結果可能會成為其自身的輸入[11]，經過combine函數處理的數據會再次進入combine函數，但輪詢問題是不可避免的，所以要保證combine函數的輸入類型和輸出類型必須一致，若不一致，要增加判斷的邏輯。

4 實驗結果分析

4.1 實驗環境

本文通過部署一個Hadoop偽分布環境，通過實驗對比和分析來驗證文中關于利用Combine優化機制理論的正確性。在實驗搭建的集群中，NameNode和DataNode都在本機中。

硬件環境：Intel? Core? i5 CPU M 480 @ 2.67GHz×4，3.7GB內存，500GB硬盤。

軟件環境：Fedora16，Hadoop 0.21.0，Jdk-6u27-linux-i586。

4.2 實驗結果及分析

本文采用數字求和程序作為實驗測試程序，采用文本文檔作為實驗數據集，文本中數據隨機生成，正負相間，以防數據全正或全負致使計算結果溢出。

⑴ 實驗一

采用418.9MB文本文檔數據集進行多次實驗，分析實驗日志，取其中三組實驗結果列表，利用Combine機制優化前后，框架中各部分的輸入輸出和耗時對比如表1所示。

作業結束后，查看日志，利用Combine機制優化前后作業都分別共執行了12個map任務和1個reduce任務，表1中詳細記錄了作業執行完成后map函數、combine函數和reduce函數的輸入輸出數量和耗時，通過表1的數據對比可以看出：一方面，使用Combine優化機制優化后，作業執行時間明顯減少；另一方面，從實驗數據中可以看到combine函數輸入的數量要明顯大于map函數輸出的數量，而且combine函數輸出的數量要明顯大于reduce函數輸入的數量，所以可以驗證在使用Combine優化過程中存在明顯的輪詢問題，但由于實驗中輸入輸出數據類型相同，所以輪詢并不影響實驗的最終輸出結果，Combiner的輸出結果被寫到中間文件，并被發送到reduce任務中，經reduce函數處理后直接被寫到最終的輸出文件中，保存在HDFS文件系統上。

⑵ 實驗二

分別采用41.9MB、83.8MB、167.6MB、335.1MB、418.9MB文本文檔作為測試數據集。

圖3為利用Combine優化機制優化前后作業執行總耗時對比，在圖3中可以看到，隨著數據集增大，利用Combine機制優化過的作業在總耗時方面時間減少明顯，性能平均提高43%，可以充分驗證使用Combine優化機制會大量減少Map端最后寫到磁盤的數據量，同時也減少了網絡中傳輸的數據量，大幅提高系統性能。

圖4為利用Combine優化機制優化前后Map端計算耗時對比，可以看到，利用Combine優化機制優化后，計算機壓力雖然大部分在Map端，但是Map端耗時隨著數據集增大而減少也很明顯。

圖5為利用Combine優化機制優化前后Reduce端耗時對比，因為數據求和的實驗在利用Combine優化機制優化后，計算壓力大部分都在Map端，所以可以看到當數據集增大時，Reduce端執行時間基本都維持在4秒左右。

5 結束語

本文通過對Shuffle流程的詳細分析，指出Combine機制在Shuffle流程中的具體執行位置，并詳細分析了Combine優化機制執行所需條件和執行過程中存在的問題。通過實驗分析，可以看到Combine機制對性能的提升確實很大，但在實際開發應用中應該結合文中所述和實際開發需要，評測是否需要利用Combine優化機制，只有在網絡帶寬資源有限，對系統的瓶頸比較大時才應該考慮使用Combine機制，以減少數據在網絡中的傳輸量，降低系統對網絡帶寬的需求。

參考文獻：

[1] Tom White.Hadoop權威指南[M].清華大學出版社，2011.

[2] 周一可.云計算下MapReduce 編程模型可用性研究與優化[D].上海交通大學，2011.

[3] Jeffrey Dean， Sanjay Ghemawat. MapReduce： Simplified Data Processing on Large Clusters[C]，2004.

[4] Wikipedia. MapReduce介紹.2012年09月22日引自http：//zh.wikipedia.org/wiki/MapReduce

[5] 王凱.MapReduce集群多用戶作業調度方法的研究與實現[D].國防科學技術大學，2010.11.

[6] 徐強，王振江.云計算應用開發實踐[M].機械工業出版社，2012.

[7] gaobotian. Hadoop源代碼分析（完整版）. 2012年09月22日引自http：//wenku.baidu.com/view/ffc10130eefdc8d376ee32ec.html

[8] 皮冰鋒等.Hadoop開發者第一期.2012年9月22日引自http：//www.hadoopor.com/

[9] Lam， C著，韓冀中譯.Hadoop實戰[M].人民郵電出版社，2011.

[10] StackOverFlow. "Combiner" Class in a mapreduce job. 2012年9月23日引自http：//stackoverflow.com/questions/10220371/Combiner-class-in-a-mapreduce-job

[11] 何忠育等.Hadoop開發者第四期.2012年9月23日引自 http：//www.hadoopor.com/