一、Combiner的出现
（1）为什么需要进行Map规约操
作
在上述过程中，我们看到至少两个性能瓶颈：
（1）如果我们有10亿个数据，Mapper会生成10亿个键值对在网络间进行传输，但如果我们只是对数据求最大值，那么很明显的Mapper只需要输出它所知道的最大值即可。这样做不仅可以减轻网络压力，同样也可以大幅度提高程序效率。
总结：网络带宽严重被占降低程序效率；
（2）假设使用美国专利数据集中的国家一项来阐述数据倾斜这个定义，这样的数据远远不是一致性的或者说平衡分布的，由于大多数专利的国家都属于美国，这样不仅Mapper中的键值对、中间阶段(shuffle)的键值对等，大多数的键值对最终会聚集于一个单一的Reducer之上，压倒这个Reducer，从而大大降低程序的性能。
总结：单一节点承载过重降低程序性能；
（2）一种方案能够解决这两个问题呢？
在MapReduce编程模型中，在Mapper和Reducer之间有一个非常重要的组件，它解决了上述的性能瓶颈问题，它就是Combiner。
①与mapper和reducer不同的是，combiner没有默认的实现，需要显式的设置在conf中才有作用。
②并不是所有的job都适用combiner，只有操作满足结合律的才可设置combiner。
combine操作类似于：opt(opt(1, 2, 3), opt(4, 5, 6))。如果opt为求和、求最大值的话，可以使用，但是如果是求中值的话，不适用。
二、Combiner的作用
当Map程序开始产生结果的时候，并不是直接写到文件的，而是利用缓存做一些排序方面的预处理操作。
每个Map任务都有一个循环内存缓冲区（默认100MB），当缓存的内容达到80%时，后台线程开始将内容写到文件，此时Map任务可以持续输出结果，但如果缓冲区满了，Map任务则需要等待。
写文件使用round-robin方式。在写入文件之前，先将数据按照Reduce进行分区。对于每一个分区，都会在内存中根据key进行排序，如果配置了Combiner，则排序后执行Combiner（Combine之后可以减少写入文件和传输的数据）。
每次结果达到缓冲区的阀值时，都会创建一个文件，在Map结束时，可能会产生大量的文件。在Map完成前，会将这些文件进行合并和排序。如果文件的数量超过3个，则合并后会再次运行Combiner（1、2个文件就没有必要了）。
（1）MapReduce的一种优化手段
每一个map都可能会产生大量的本地输出，Combiner的作用就是对map端的输出先做一次合并，以减少在map和reduce节点之间的数据传输量，以提高网络IO性能
（2）Combiner的过程
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）
例如wordcount的例子和找出value的最大值的程序
combiner和reduce完全一致，如下所示：
map: (K1, V1) → list(K2, V2) combine: (K2, list(V2)) → list(K3, V3) reduce: (K3, list(V3)) → list(K4, V4)
使用combiner之后，先完成的map会在本地聚合，提升速度。对于hadoop自带的wordcount的例子，value就是一个叠加的数字，所以map一结束就可以进行reduce的value叠加，而不必要等到所有的map结束再去进行reduce的value叠加。
（3）融合Combiner的MapReduce
1）使用MyReducer作为Combiner
// 设置Map规约Combiner
job.setCombinerClass(MyReducer.class);
执行后看到map的输出和combine的输入统计是一致的，而combine的输出与reduce的输入统计是一样的。
由此可以看出规约操作成功，而且执行在map的最后，reduce之前。
2）自己定义Combiner
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）添加设置Combiner的代码
// 设置Map规约Combiner
job.setCombinerClass(MyCombiner.class);
小结： 在实际的Hadoop集群操作中，我们是由多台主机一起进行MapReduce的， 如果加入规约操作，每一台主机会在reduce之前进行一次对本机数据的规约， 然后在通过集群进行reduce操作，这样就会大大节省reduce的时间， 从而加快MapReduce的处理速度
