ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > Hadoop > 一个MapReduce 程序示例 细节决定成败(三) ：Combiner

一个MapReduce 程序示例 细节决定成败(三) ：Combiner

原创 Hadoop 作者：self_control 时间：2016-05-27 13:09:27 0 删除 编辑
上一篇中，我们写了一个统计输入文件中 a~z 每个字符个数的mr 程序。通过查看运行日志的Counter 计数可以看到map 到 reduce 之间的网络传输是多少。
而本篇我们将介绍的Combiner 是一个非常重要的组件，主要可以用来减少网络传输。原理是在每个map 任务运行节点上，先把map的输出进行汇总，然后再传输到
reducer任务，也可以称此为一个map 端的reduce操作。
先上代码。因为我们的combiner 逻辑与 reducer逻辑相同，所以我们简单的使用MyWordCountReducer来设置成任务的combiner 。

点击(此处)折叠或打开

  1. job.setCombinerClass(MyWordCountJob.MyWordCountReducer.class);

重点看运行日志：

点击(此处)折叠或打开

  1. File System Counters
  2.                 FILE: Number of bytes read=422
  3.                 FILE: Number of bytes written=338601
  4.                 FILE: Number of read operations=0
  5.                 FILE: Number of large read operations=0
  6.                 FILE: Number of write operations=0
  7.                 HDFS: Number of bytes read=556
  8.                 HDFS: Number of bytes written=103
  9.                 HDFS: Number of read operations=12
  10.                 HDFS: Number of large read operations=0
  11.                 HDFS: Number of write operations=2
  12.         Job Counters
  13.                 Launched map tasks=3
  14.                 Launched reduce tasks=1
  15.                 Data-local map tasks=3
  16.                 Total time spent by all maps in occupied slots (ms)=355336
  17.                 Total time spent by all reduces in occupied slots (ms)=66400
  18.         Map-Reduce Framework
  19.                 Map input records=8
  20.                 Map output records=137
  21.                 Map output bytes=822
  22.                 Map output materialized bytes=434
  23.                 Input split bytes=399
  24.                 Combine input records=137
  25.                 Combine output records=52
  26.                 Reduce input groups=25
  27.                 Reduce shuffle bytes=434
  28.                 Reduce input records=52
  29.                 Reduce output records=25
  30.                 Spilled Records=104
  31.                 Shuffled Maps =3
  32.                 Failed Shuffles=0
  33.                 Merged Map outputs=3
  34.                 GC time elapsed (ms)=274
  35.                 CPU time spent (ms)=3430
  36.                 Physical memory (bytes) snapshot=1078874112
  37.                 Virtual memory (bytes) snapshot=3947868160
  38.                 Total committed heap usage (bytes)=884539392
  39.         Shuffle Errors
  40.                 BAD_ID=0
  41.                 CONNECTION=0
  42.                 IO_ERROR=0
  43.                 WRONG_LENGTH=0
  44.                 WRONG_MAP=0
  45.                 WRONG_REDUCE=0
  46.         File Input Format Counters
  47.                 Bytes Read=157
  48.         File Output Format Counters
  49.                 Bytes Written=103
通过日志可以看出， map 的输入输出记录数为8和137 这个与没combiner没影响。但我们发现reduce的输入由137变成了52。也就是说通过网络传输的数据量减少了。
同时也可以发现这个变化是combine 影响的。 可以看到 map 输出的137条记录实际是进行combine中处理了，combine 输入为map输入的137条记录，combine  输出的52条记录进行到reduce中继续汇总。
注意：combine 是在map 端的汇总，没有网络传输，它汇总的也只是combiner所在节点的map的输入，不是全部节点的数据。
reduce 中处理的才是所有节点的数据。
下一篇，我们介绍另一种类似combiner功能的实现方式。in-map 聚合。

最后还是把全部代码贴到后面吧


点击(此处)折叠或打开

  1. package wordcount;

  3. import java.io.IOException;

  5. import org.apache.commons.lang.StringUtils;
  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.conf.Configured;
  8. import org.apache.hadoop.fs.Path;
  9. import org.apache.hadoop.io.IntWritable;
  10. import org.apache.hadoop.io.LongWritable;
  11. import org.apache.hadoop.io.Text;
  12. import org.apache.hadoop.mapreduce.Job;
  13. import org.apache.hadoop.mapreduce.Mapper;
  14. import org.apache.hadoop.mapreduce.Reducer;
  15. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  16. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  17. import org.apache.hadoop.util.Tool;
  18. import org.apache.hadoop.util.ToolRunner;
  19. import org.apache.log4j.Logger;

  21. public class MyWordCountJob extends Configured implements Tool {
  22.         Logger log = Logger.getLogger(MyWordCountJob.class);

  24.         public static class MyWordCountMapper extends
  25.                         Mapper<LongWritable, Text, Text, IntWritable> {
  26.                 Logger log = Logger.getLogger(MyWordCountJob.class);

  28.                 Text mapKey = new Text();
  29.                 IntWritable mapValue = new IntWritable(1);
  30.                 @Override
  31.                 protected void map(LongWritable key, Text value, Context context)
  32.                                 throws IOException, InterruptedException {
  33.                         for(char c :value.toString().toLowerCase().toCharArray()){
  34.                                 if(c>='a' && c <='z'){
  35.                                         mapKey.set(String.valueOf(c));
  36.                                         context.write(mapKey, mapValue);
  37.                                 }
  38.                         }
  39.                 }

  41.         }


  44.         public static class MyWordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
  45.                 Text rkey = new Text();
  46.                 IntWritable rvalue = new IntWritable(1);
  47.                 @Override
  48.                 protected void reduce(Text key, Iterable<IntWritable> values,Context context)
  49.                                 throws IOException, InterruptedException {
  50.                         int n=0;
  51.                         for(IntWritable value :values){
  52.                                 n+= value.get();
  53.                         }
  54.                         rvalue.set(n);
  55.                         context.write(key, rvalue);
  56.                 }
  57.         }

  59.         @Override
  60.         public int run(String[] args) throws Exception {
  61.                 //valid the parameters
  62.                 if(args.length !=2){
  63.                         return -1;
  64.                 }

  66.                 Job job = Job.getInstance(getConf(), "MyWordCountJob");
  67.                 job.setJarByClass(MyWordCountJob.class);

  69.                 Path inPath = new Path(args[0]);
  70.                 Path outPath = new Path(args[1]);

  72.                 outPath.getFileSystem(getConf()).delete(outPath,true);
  73.                 TextInputFormat.setInputPaths(job, inPath);
  74.                 TextOutputFormat.setOutputPath(job, outPath);


  77.             job.setMapperClass(MyWordCountJob.MyWordCountMapper.class);
  78.                 job.setReducerClass(MyWordCountJob.MyWordCountReducer.class);
  79.                 job.setCombinerClass(MyWordCountJob.MyWordCountReducer.class);
  80.             job.setInputFormatClass(TextInputFormat.class);
  81.             job.setOutputFormatClass(TextOutputFormat.class);

  83.                 job.setMapOutputKeyClass(Text.class);
  84.                 job.setMapOutputValueClass(IntWritable.class);
  85.                 job.setOutputKeyClass(Text.class);
  86.                 job.setOutputValueClass(IntWritable.class);

  88.                 return job.waitForCompletion(true)?0:1;
  89.         }
  90.         public static void main(String [] args){
  91.                 int result = 0;
  92.                 try {
  93.                         result = ToolRunner.run(new Configuration(), new MyWordCountJob(), args);
  94.                 } catch (Exception e) {
  95.                         e.printStackTrace();
  96.                 }
  97.                 System.exit(result);
  98.         }

  100. }
下一篇中，我们使用in-map Aggregation来优化我们的任务。在某种情况下，可以比combiner更有效率。 
一个MapReduce 程序示例 细节决定成败(四) ：In-Map Aggregation 

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/30066956/viewspace-2107885/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： 一个MapReduce 程序示例 细节决定成败(二) ：观察日志及 Counter
下一篇： 一个MapReduce 程序示例 细节决定成败(四) ：In-Map Aggregation
请登录后发表评论 登录
全部评论
self_control

注册时间：2014-12-16

  • 博文量
    49
  • 访问量
    406895

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593