一、Combiner的出現背景

1.1 回顧Map階段五大步驟

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

map section

　　其中，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規約Combinerjob.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 extendsMapper<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 extendsReducer<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 extendsReducer<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規約Combinerjob.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的處理速度。