MapReduce之Combiner详解
程序员文章站
2022-06-30 11:47:00
...
Combiner是什么?为什么会出现Combiner?
Combiner是一个本地化的reduce操作,它是map运算的后续操作,
主要是在map计算出中间文件前做一个简单的合并重复key值的操作
1、网络带宽严重被占降低程序效率;
(提前在map上执行分组,减少传输给reduce的数据量)
2、单一节点承载过重降低程序性能;
(全在ruduce上运行,导致负载过重)
每一个map都可能会产生大量的本地输出,Combiner的作用就是对map端的输出先做一次合并,
以减少在map和reduce节点之间的数据传输量,以提高网络IO性能
例如:对于hadoop自带的wordcount的例子,value就是一个叠加的数字,
所以map一结束就可以进行reduce的value叠加,而不必要等到所有的map结束再去进行reduce的value叠加。
Combiner设置
// 设置Map规约Combiner
job.setCombinerClass(MyReducer.class);
执行后看到map的输出和combine的输入统计是一致的,而combine的输出与reduce的输入统计是一样的。
由此可以看出规约操作成功,而且执行在map的最后,reduce之前。
Combiner限制
①与mapper和reducer不同的是,combiner没有默认的实现,需要显式的设置在conf中才有作用。
②并不是所有的job都适用combiner,只有操作满足结合律的才可设置combiner。
combine操作类似于:opt(opt(1, 2, 3), opt(4, 5, 6))。如果opt为求和、求最大值的话,
可以使用,但是如果是求中值的话,不适用。
(因为任务分片后,每个map保存有一定数据,如果要提前在map上操作,那么map数据间不能有依赖关系)
小结:
在实际的Hadoop集群操作中,我们是由多台主机一起进行MapReduce的,
如果加入规约操作,每一台主机会在reduce之前进行一次对本机数据的规约,
然后在通过集群进行reduce操作,这样就会大大节省reduce的时间,
从而加快MapReduce的处理速度
自定义Combiner:
public static class MyCombiner extends Reducer<Text, LongWritable, Text, LongWritable> {
protected void reduce(
Text key, Iterable<LongWritable> values,Context context)throws 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 + ">");
};
}