解析

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

Combiner 和 Reducer 一样，编写一个类，然后继承 Reducer，reduce 方法中写具体的 Combiner 逻辑，然后在 job 中设置 Combiner 组

执行后看到map的输出和combine的输入统计是一致的，而combine的输出与reduce的输入统计是一样的。由此可以看出规约操作成功，而且执行在map的最后，reduce之前。

其实Combiner也可以看成就是Reduce，俩者的代码完全一样

Combiner注意点

有很多人认为这个combiner和map输出的数据合并是一个过程，其实不然，map输出的数据合并只会产生在有数据spill出的时候，即进行merge操作。

与mapper与reducer不同的是，combiner没有默认的实现，需要显式的设置在conf中才有作用。

并不是所有的job都适用combiner，只有操作满足结合律的才可设置combiner。combine操作类似于：opt(opt(1, 2, 3), opt(4, 5, 6))。如果opt为求和、求最大值的话，可以使用，但是如果是求中值的话，不适用。

代码实现

我们以WordCount为例，先是没有使用Combiner的

Mapper

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class MapTest extends Mapper<LongWritable, Text, Text, IntWritable> {Text k = new Text();IntWritable v = new IntWritable(1);@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String words[] = value.toString().split(" ");for (String word : words) {k.set(word);context.write(k, v);}}
}

Reduce

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class RedTest extends Reducer<Text, IntWritable, Text, IntWritable> {IntWritable v = new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int sum = 0;for (IntWritable value : values) {sum += 1;}v.set(sum);context.write(key, v);}
}

Driver

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class DriTest {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(DriTest.class);job.setMapperClass(MapTest.class);job.setReducerClass(RedTest.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);FileInputFormat.setInputPaths(job, "D:\\MP\\WordCount\\input");Path path = new Path("D:\\MP\\WordCount\\output");if (path.getFileSystem(conf).exists(path)) {path.getFileSystem(conf).delete(path, true);}FileOutputFormat.setOutputPath(job,path);boolean b = job.waitForCompletion(true);System.exit(b ? 0 : 1);}
}

运行日志

第一个是Mapper的输出
第二个是Combiner的，发现都是0
第三个是Reduce接收的
在这里插入图片描述

加上Combiner

我们在Driver阶段加上一句
job.setCombinerClass(RedTest.class);
运行查看日志
在这里插入图片描述
我们发现第一个是没有任何变化的
第二个不再是0了，而是有了数据
第三的明显比没有加Combiner的少了许多，这就说明Combiner生效了，起到了为Reduce减负的作用