一、Combiner的出現背景

1.1 回顧Map階段五大步驟

　　在第四篇博文《初識MapReduce》中，我們認識了MapReduce的八大步湊，其中在Map階段總共五個步驟，如下圖所示：

　　其中，step1.5是一個可選步驟，它就是我們今天需要了解的 Map規約 階段。現在，我們再來看看前一篇博文《計數器與自定義計數器》中的第一張關於計數器的圖：

　　我們可以發現，其中有兩個計數器：Combine output records和Combine input records，他們的計數都是0，這是因為我們在程式碼中沒有進行Map階段的規約操作。

1.2 為什麼需要進行Map規約操作

　　眾所周知，Hadoop框架使用Mapper將資料處理成一個個的<key,value>鍵值對，在網路節點間對其進行整理(shuffle)，然後使用Reducer處理資料並進行最終輸出。

　　在上述過程中，我們看到至少兩個效能瓶頸：

　　（1）如果我們有10億個資料，Mapper會生成10億個鍵值對在網路間進行傳輸，但如果我們只是對資料求最大值，那麼很明顯的Mapper只需要輸出它所知道的最大值即可。這樣做不僅可以減輕網路壓力，同樣也可以大幅度提高程式效率。

　　總結：網路頻寬嚴重被佔降低程式效率；

　　（2）假設使用美國專利資料集中的國家一項來闡述資料傾斜這個定義，這樣的資料遠遠不是一致性的或者說平衡分佈的，由於大多數專利的國家都屬於美國，這樣不僅Mapper中的鍵值對、中間階段(shuffle)的鍵值對等，大多數的鍵值對最終會聚集於一個單一的Reducer之上，壓倒這個Reducer，從而大大降低程式的效能。

　　總結：單一節點承載過重降低程式效能；

那麼，有木有一種方案能夠解決這兩個問題呢？

二、初步探索Combiner

2.1 Combiner的橫空出世

在MapReduce程式設計模型中，在Mapper和Reducer之間有一個非常重要的元件，它解決了上述的效能瓶頸問題，它就是Combiner。

PS：

①與mapper和reducer不同的是，combiner沒有預設的實現，需要顯式的設定在conf中才有作用。

②並不是所有的job都適用combiner，只有操作滿足結合律的才可設定combiner。combine操作類似於：opt(opt(1, 2, 3), opt(4, 5, 6))。如果opt為求和、求最大值的話，可以使用，但是如果是求中值的話，不適用。

　　每一個map都可能會產生大量的本地輸出，Combiner的作用就是對map端的輸出先做一次合併，以減少在map和reduce節點之間的資料傳輸量，以提高網路IO效能，是MapReduce的一種優化手段之一，其具體的作用如下所述。

　　（1）Combiner最基本是實現本地key的聚合，對map輸出的key排序，value進行迭代。如下所示：

　　map: (K1, V1) → list(K2, V2) 
　　combine: (K2, list(V2)) → list(K2, V2) 
　　reduce: (K2, list(V2)) → list(K3, V3)

　　（2）Combiner還有本地reduce功能（其本質上就是一個reduce），例如Hadoop自帶的wordcount的例子和找出value的最大值的程式，combiner和reduce完全一致，如下所示：

　　map: (K1, V1) → list(K2, V2) 
　　combine: (K2, list(V2)) → list(K3, V3) 
　　reduce: (K3, list(V3)) → list(K4, V4)

PS：現在想想，如果在wordcount中不用combiner，那麼所有的結果都是reduce完成，效率會相對低下。使用combiner之後，先完成的map會在本地聚合，提升速度。對於hadoop自帶的wordcount的例子，value就是一個疊加的數字，所以map一結束就可以進行reduce的value疊加，而不必要等到所有的map結束再去進行reduce的value疊加。

2.2 融合Combiner的MapReduce

　　前面文章中的程式碼都忽略了一個可以優化MapReduce作業所使用頻寬的步驟—Combiner，它在Mapper之後Reducer之前執行。Combiner是可選的，如果這個過程適合於你的作業，Combiner例項會在每一個執行map任務的節點上執行。Combiner會接收特定節點上的Mapper例項的輸出作為輸入，接著Combiner的輸出會被髮送到Reducer那裡，而不是傳送Mapper的輸出。Combiner是一個“迷你reduce”過程，它只處理單臺機器生成的資料。

2.3 使用MyReducer作為Combiner

在前面文章中的WordCount程式碼中加入以下一句簡單的程式碼，即可加入Combiner方法：

    // 設定Map規約Combiner
    job.setCombinerClass(MyReducer.class);

　　還是以下面的檔案內容為例，看看這次計數器會發生怎樣的改變？

　　（1）上傳的測試檔案的內容

hello edison
hello kevin

　　（2）除錯後的計數器日誌資訊

　　可以看到，原本都為0的Combine input records和Combine output records發生了改變。我們可以清楚地看到map的輸出和combine的輸入統計是一致的，而combine的輸出與reduce的輸入統計是一樣的。由此可以看出規約操作成功，而且執行在map的最後，reduce之前。

三、自己定義Combiner

　　為了能夠更加清晰的理解Combiner的工作原理，我們自定義一個Combiners類，不再使用MyReduce做為Combiners的類，具體的程式碼下面一一道來。

3.1 改寫Mapper類的map方法

　　public static class MyMapper extends
            Mapper<LongWritable, Text, Text, LongWritable> {
        protected void map(LongWritable key, Text value,
                Mapper<LongWritable, Text, Text, LongWritable>.Context context)
                throws java.io.IOException, InterruptedException {
            String line = value.toString();
            String[] spilted = line.split(" ");
            for (String word : spilted) {
                context.write(new Text(word), new LongWritable(1L));
                // 為了顯示效果而輸出Mapper的輸出鍵值對資訊
                System.out.println("Mapper輸出<" + word + "," + 1 + ">");
            }
        };
    }

3.2 改寫Reducer類的reduce方法

     public static class MyReducer extends
            Reducer<Text, LongWritable, Text, LongWritable> {
        protected void reduce(Text key,
                java.lang.Iterable<LongWritable> values,
                Reducer<Text, LongWritable, Text, LongWritable>.Context context)
                throws java.io.IOException, InterruptedException {
            // 顯示次數表示redcue函式被呼叫了多少次，表示k2有多少個分組
            System.out.println("Reducer輸入分組<" + key.toString() + ",N(N>=1)>");

            long count = 0L;
            for (LongWritable value : values) {
                count += value.get();
                // 顯示次數表示輸入的k2,v2的鍵值對數量
                System.out.println("Reducer輸入鍵值對<" + key.toString() + ","
                        + value.get() + ">");
            }
            context.write(key, new LongWritable(count));
        };
    }

3.3 新增MyCombiner類並重寫reduce方法

    public static class MyCombiner extends
            Reducer<Text, LongWritable, Text, LongWritable> {
        protected void reduce(
                Text key,
                java.lang.Iterable<LongWritable> values,
                org.apache.hadoop.mapreduce.Reducer<Text, LongWritable, Text, LongWritable>.Context context)
                throws java.io.IOException, InterruptedException {
            // 顯示次數表示規約函式被呼叫了多少次，表示k2有多少個分組
            System.out.println("Combiner輸入分組<" + key.toString() + ",N(N>=1)>");
            long count = 0L;
            for (LongWritable value : values) {
                count += value.get();
                // 顯示次數表示輸入的k2,v2的鍵值對數量
                System.out.println("Combiner輸入鍵值對<" + key.toString() + ","
                        + value.get() + ">");
            }
            context.write(key, new LongWritable(count));
            // 顯示次數表示輸出的k2,v2的鍵值對數量
            System.out.println("Combiner輸出鍵值對<" + key.toString() + "," + count
                    + ">");
        };
    }

3.4 新增設定Combiner的程式碼

    // 設定Map規約Combiner
    job.setCombinerClass(MyCombiner.class);

3.5 除錯執行的控制檯輸出資訊

　　（1）Mapper

Mapper輸出<hello,1>
Mapper輸出<edison,1>
Mapper輸出<hello,1>
Mapper輸出<kevin,1>

　　（2）Combiner

Combiner輸入分組<edison,N(N>=1)>
Combiner輸入鍵值對<edison,1>
Combiner輸出鍵值對<edison,1>
Combiner輸入分組<hello,N(N>=1)>
Combiner輸入鍵值對<hello,1>
Combiner輸入鍵值對<hello,1>
Combiner輸出鍵值對<hello,2>
Combiner輸入分組<kevin,N(N>=1)>
Combiner輸入鍵值對<kevin,1>
Combiner輸出鍵值對<kevin,1>

　　這裡可以看出，在Combiner中進行了一次本地的Reduce操作，從而簡化了遠端Reduce節點的歸併壓力。

　　（3）Reducer

Reducer輸入分組<edison,N(N>=1)>
Reducer輸入鍵值對<edison,1>
Reducer輸入分組<hello,N(N>=1)>
Reducer輸入鍵值對<hello,2>
Reducer輸入分組<kevin,N(N>=1)>
Reducer輸入鍵值對<kevin,1>

　　這裡可以看出，在對hello的歸併上，只進行了一次操作就完成了。

　　那麼，如果我們再來看看不新增Combiner時的控制檯輸出資訊：

　　（1）Mapper

Mapper輸出<hello,1>
Mapper輸出<edison,1>
Mapper輸出<hello,1>
Mapper輸出<kevin,1>

　　（2）Reducer

Reducer輸入分組<edison,N(N>=1)>
Reducer輸入鍵值對<edison,1>
Reducer輸入分組<hello,N(N>=1)>
Reducer輸入鍵值對<hello,1>
Reducer輸入鍵值對<hello,1>
Reducer輸入分組<kevin,N(N>=1)>
Reducer輸入鍵值對<kevin,1>

　　可以看出，沒有采用Combiner時hello都是由Reducer節點來進行統一的歸併，也就是這裡為何會有兩次hello的輸入鍵值對了。

總結：從控制檯的輸出資訊我們可以發現，其實combine只是把兩個相同的hello進行規約，由此輸入給reduce的就變成了<hello,2>。在實際的Hadoop叢集操作中，我們是由多臺主機一起進行MapReduce的，如果加入規約操作，每一臺主機會在reduce之前進行一次對本機資料的規約，然後在通過叢集進行reduce操作，這樣就會大大節省reduce的時間，從而加快MapReduce的處理速度。

參考資料

（4）iPolaris，《Hadoop中Combiner的使用》：http://blog.csdn.net/ipolaris/article/details/8723782

作者：周旭龍

本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連結。