上一篇文章我们已经对shuffle阶段学习了分区和排序，现在来学习下规约。每一个map都可能善生大量的输出，conbiner的作用就是对map端的输出先做一次合并，以减少在map和reduce节点之间的数据传输量，以提高IO性能，是mapReduce的一种优化手段之一。就是会把相同的键值进行合并，这就是所谓的规约

  • combiner是MR程序中Mapper和Reducer之外的一个组件
  • combiner组件的父类就是Reducer

  • combiner和reducer的区别在于运行的位置

    1. combiner是在每一个mapTask所在的节点运行
      2. Reducer是接收全局所有Mapper的输出结果

  • combiner的意义就是对每一个mapTask的输出进行局部汇总，减少网络IO传输

    实现步骤

  1. 自定义一个combiner集成Reducer，重写reduce方法
  2. 在job设置job.setCombinerClass(CustomCombiner.class)

combiner能够应用的前提是不能影响最终的业务逻辑，而且combiner的输出kv应该跟reducer的输入kv类型要对应起来.

案例

还是拿我们统计单词出现的次数举例：案例示例地址
我们要做的仅仅是在map到reduce阶段增加一个规约处理，以此来减少网络传输量，来提高执行效率。
image.png


代码实现

WordCountMapper 不做改变

  1. package com.dongnaoedu.network.hadoop.mapreduce.combiner;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Mapper;
  7. import java.io.IOException;
  9. public class WordCountMapper extends Mapper<LongWritable, Text,Text,LongWritable> {
  10.     @Override
  11.     public void map(LongWritable key,Text value,Context context) throws IOException,InterruptedException{
  12.         String line = value.toString();
  13.         String[] split = line.split(",");
  14.         for(String word : split){
  15.             context.write(new Text(word), new LongWritable(1));
  16.         }
  17.     }
  18. }

增加的MyCombiner

  1. package com.dongnaoedu.network.hadoop.mapreduce.combiner;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Reducer;
  7. import java.io.IOException;
  9. /**
  10.  * @author heian
  11.  * @date 2022/3/24 12:25 下午
  12.  * @description 规约
  13.  */
  14. public class MyCombiner extends Reducer<Text, LongWritable, Text,LongWritable> {
  16.     /**
  17.      *  自定义我们的reduce逻辑
  18.      *  所有key都是我们的单词，所有的values都是我们单词出现的次数
  19.      */
  20.     @Override
  21.     protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException,InterruptedException{
  22.         long count = 0;
  23.         for(LongWritable value : values){
  24.             count += value.get();
  25.         }
  26.         context.write(key,new LongWritable(count));
  27.     }
  29. }

SortReducer 不做改变

  1. package com.dongnaoedu.network.hadoop.mapreduce.combiner;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Reducer;
  7. import java.io.IOException;
  9. public class WordCountReduce extends Reducer<Text,LongWritable, Text,LongWritable> {
  10.     @Override
  11.     protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException,InterruptedException{
  12.         long count = 0;
  13.         // 此时集合经过规约处理，元素只有一个而已
  14.         for(LongWritable value : values){
  15.             count += value.get();
  16.         }
  17.         context.write(key,new LongWritable(count));
  18.     }
  19. }

设置JobMain的规约类

  1. package com.dongnaoedu.network.hadoop.mapreduce.combiner;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.conf.Configured;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.LongWritable;
  7. import org.apache.hadoop.io.Text;
  8. import org.apache.hadoop.mapreduce.Job;
  9. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  10. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  11. import org.apache.hadoop.util.Tool;
  12. import org.apache.hadoop.util.ToolRunner;
  15. public class JobMain extends Configured implements Tool{
  17.     @Override
  18.     public int run(String[] strings) throws Exception{
  19.         Job job = Job.getInstance(super.getConf(), "mapReduce_combiner");
  21.         //打包放在集群下运行，需要做一个配置
  22.         job.setJarByClass(JobMain.class);
  24.         // 第一步：设置读取文件的类：k1 v1
  25.         job.setInputFormatClass(TextInputFormat.class);
  26.         TextInputFormat.addInputPath(job, new Path("hdfs://node01:8082/wordcount"));
  28.         // 第二步：设置mapper类
  29.         job.setMapperClass(WordCountMapper.class);
  30.         // 设置map阶段输出类型
  31.         job.setMapOutputKeyClass(Text.class);
  32.         job.setMapOutputValueClass(LongWritable.class);
  34.         // 第三 四 五 六步采用默认方式（分区 排序 规约 分组）
  35.         job.setCombinerClass(MyCombiner.class);
  37.         // 第七步:设置Reduce类
  38.         job.setReducerClass(WordCountReduce.class);
  39.         job.setOutputKeyClass(Text.class);
  40.         job.setOutputValueClass(LongWritable.class);
  42.         // 第八步：设置输出类
  43.         job.setOutputFormatClass(TextOutputFormat.class);
  44.         // 设置输出路径
  45.         TextOutputFormat.setOutputPath(job, new Path("hafs://node01:8082/wordcount_cout"));
  47.         boolean b = job.waitForCompletion(true);
  49.         return b ? 0:1;
  50.     }
  52.     public static void main(String[] args) throws Exception{
  53.         Configuration configuration = new Configuration();
  54.         // 启动一个任务  run=0成功
  55.         int run = ToolRunner.run(configuration, new JobMain(), args);
  56.         System.exit(run);  
  57.     }
  59. }