Hadoop学习笔记—8.Combiner与自定义Combiner
一、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規約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的處理速度。
參考資料
（1）萬川梅、謝正蘭，《Hadoop應用開發實戰詳解（修訂版）》：http://item.jd.com/11508248.html
（2）Suddenly，《Hadoop日記Day17-計數器、map規約與分區學習》：http://www.cnblogs.com/sunddenly/p/4009568.html
（3）guoery，《MapReduce中Combiner的使用及誤區》：http://blog.csdn.net/guoery/article/details/8529004
（4）iPolaris，《Hadoop中Combiner的使用》：http://blog.csdn.net/ipolaris/article/details/8723782
?
原文鏈接：http://edisonchou.cnblogs.com/
轉載于:https://www.cnblogs.com/zzmmyy/p/7777286.html
總結
以上是生活随笔為你收集整理的Hadoop学习笔记—8.Combiner与自定义Combiner的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 第六讲：PrintClient工具的使用
- 下一篇： phantomjs