码迷，mamicode.com
首页 > 其他好文 > 详细

Hadoop Combiner组件

时间：2016-01-31 21:13:04      阅读：259      评论：0      收藏：0      [点我收藏+]

标签：

一：背景

在MapReduce模型中，reduce的功能大多是统计分类类型的总量、求最大值最小值等，对于这些操作可以考虑在Map输出后进行Combiner操作，这样可以减少网络传输负载，同时减轻reduce任务的负担。Combiner操作是运行在每个节点上的，只会影响本地Map的输出结果，Combiner的输入为本地map的输出结果，很多时候Combiner的逻辑和reduce的逻辑是相同的，因此两者可以共用reducer体。

 

二：什么时候运行Combiner

(1)：当job设置了Combiner，并且spill的个数达到了min.num.spill.for.combine(默认是3)的时候，那么Combiner就会在merge之前执行。

(2)：但是有的情况下，merge开始执行，但spill文件的个数没有达到需求，这个时候Combiner可能会在merge之后运行。

(3)：Combiner也有可能不运行，Combiner会考虑当时集群的一个负载情况。

 

 

三：程序代码

 

[java] view plain copy
 
  1. public class WordCountTest {  
  2.   
  3.     // 定义输入路径  
  4.     private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/hello";  
  5.     // 定义输出路径  
  6.     private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";  
  7.   
  8.     public static void main(String[] args) {  
  9.   
  10.         try {  
  11.             // 创建配置信息  
  12.             Configuration conf = new Configuration();  
  13.               
  14.             /**********************************************/  
  15.             //对Map端输出进行压缩  
  16.             //conf.setBoolean("mapred.compress.map.output", true);  
  17.             //设置map端输出使用的压缩类  
  18.             //conf.setClass("mapred.map.output.compression.codec", GzipCodec.class, CompressionCodec.class);  
  19.             //对reduce端输出进行压缩  
  20.             //conf.setBoolean("mapred.output.compress", true);  
  21.             //设置reduce端输出使用的压缩类  
  22.             //conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);  
  23.             // 添加配置文件(我们可以在编程的时候动态配置信息，而不需要手动去改变集群)  
  24.             /* 
  25.              * conf.addResource("classpath://hadoop/core-site.xml");  
  26.              * conf.addResource("classpath://hadoop/hdfs-site.xml"); 
  27.              * conf.addResource("classpath://hadoop/hdfs-site.xml"); 
  28.              */  
  29.   
  30.             // 创建文件系统  
  31.             FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
  32.             // 如果输出目录存在，我们就删除  
  33.             if (fileSystem.exists(new Path(OUT_PATH))) {  
  34.                 fileSystem.delete(new Path(OUT_PATH), true);  
  35.             }  
  36.   
  37.             // 创建任务  
  38.             Job job = new Job(conf, WordCountTest.class.getName());  
  39.   
  40.             //1.1   设置输入目录和设置输入数据格式化的类  
  41.             FileInputFormat.setInputPaths(job, INPUT_PATH);  
  42.             job.setInputFormatClass(TextInputFormat.class);  
  43.   
  44.             //1.2   设置自定义Mapper类和设置map函数输出数据的key和value的类型  
  45.             job.setMapperClass(MyMapper.class);  
  46.               
  47.             job.setMapOutputKeyClass(Text.class);  
  48.             job.setMapOutputValueClass(LongWritable.class);  
  49.   
  50.             //1.3   设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)  
  51.             job.setPartitionerClass(HashPartitioner.class);  
  52.             job.setNumReduceTasks(1);  
  53.   
  54.             // 1.4  排序、分组  
  55.             //1.5   归约(Combiner可以和Reducer共用)  
  56.             job.setCombinerClass(MyReducer.class);  
  57.             //2.1   Shuffle把数据从Map端拷贝到Reduce端。  
  58.             //2.2   指定Reducer类和输出key和value的类型  
  59.             job.setReducerClass(MyReducer.class);  
  60.             job.setOutputKeyClass(Text.class);  
  61.             job.setOutputValueClass(LongWritable.class);  
  62.   
  63.             //2.3   指定输出的路径和设置输出的格式化类  
  64.             FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
  65.             job.setOutputFormatClass(TextOutputFormat.class);  
  66.   
  67.   
  68.             // 提交作业 退出  
  69.             System.exit(job.waitForCompletion(true) ? 0 : 1);  
  70.           
  71.         } catch (Exception e) {  
  72.             e.printStackTrace();  
  73.         }  
  74.     }  
  75.   
  76.     public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {  
  77.   
  78.         // 定义一个LongWritable对象作为map输出的value类型  
  79.         LongWritable oneTime = new LongWritable(1);  
  80.         // 定义一个Text对象作为map输出的key类型  
  81.         Text word = new Text();  
  82.   
  83.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException,  
  84.                 InterruptedException {  
  85.   
  86.             // 对每一行记录采用制表符(\t)进行分割  
  87.             String[] splits = value.toString().split("\t");  
  88.   
  89.             // 遍历字符串数组输出每一个单词  
  90.             for (String str : splits) {  
  91.   
  92.                 // 设置word  
  93.                 word.set(str);  
  94.                 // 把结果写出去  
  95.                 context.write(word, oneTime);  
  96.             }  
  97.         }  
  98.     }  
  99.   
  100.     public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {  
  101.   
  102.         // 定义LongWritable对象最为Reduce输出的value类型  
  103.         LongWritable result = new LongWritable();  
  104.   
  105.         protected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException,  
  106.                 InterruptedException {  
  107.   
  108.             int sum = 0;  
  109.   
  110.             // 遍历集合，计算每个单词出现的和  
  111.             for (LongWritable s : values) {  
  112.   
  113.                 sum += s.get();  
  114.             }  
  115.             // 设置result  
  116.             result.set(sum);  
  117.             // 把结果写出去  
  118.             context.write(key, result);  
  119.         }  
  120.     }  
  121. }  



 

程序使用的数据：

技术分享

程序运行流程：

技术分享

(1)：Map的输入记录为3即<0,hello you> <10,hello me> <19,you me love>

(2)：Map的输出记录为7即<hello,1> <you,1> <hello,1> <me,1> <you,1> <me,1> <love,1>

(3)：排序分组后的记录为<hello,{1,1}> <love,{1}> <me,{1,1}>  <you,{1}> 分了4组，但是记录数依然是7，只不过是分了组而已。

(4)：进入Combiner的记录为7，经过Combiner之后的结果为<hello,2> <love,1> <me,2> <you,1>即Combiner的输出为4条记录

 

四：使用Combiner的限制

并不是所有情况都能使用Combiner，Combiner适用的场景是汇总求和、求最值的场景，但是对于求平均数的场景就不适用了。因为如果在求平均数的程序中使用了Combiner即在每个Map后都使用Combiner进行求平均，每个map计算出的平均值到了reduce端再进行平均，结果和正真的平均数就有出入了。

Hadoop Combiner组件

标签：

原文地址：http://www.cnblogs.com/thinkpad/p/5173732.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！