mapreduce的规约（Combiner）

港控/mmm° 2022-08-25 12:51 180阅读 0赞

听了超哥的一席课后逐渐明白了Combiner，记录一下自己的理解！（thanks 超哥）

首先贴上两段代码：

code1:

package combine;
    
    import java.io.IOException;
    import java.net.URI;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Counter;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    /**
     * @ClassName: WordCount2 
     * @Description: TODO(这里用一句话描述这个类的作用) 
     * @author zhangweixiang
     * @date 2014年3月6日 下午1:34:50
     */
    
    public class WordCount2 {
    	
    	static final String INPUT_PATH="hdfs://192.168.0.9:9000/hello.txt";
    	static final String OUT_PATH="hdfs://192.168.0.9:9000/word";
    	
    	public static void main(String[] args) throws Exception {
    		Configuration conf=new Configuration();
    		
    		Job job=new Job(conf, WordCount2.class.getSimpleName());
    		
    		//1.1指定读取的文件位置
    		FileInputFormat.addInputPaths(job, INPUT_PATH);
    		//指定如何对输入文件进行格式化，把输入文件每一行解析成键值对
    		job.setInputFormatClass(TextInputFormat.class);
    		
    		//1.2指定自定义的map类
    		job.setMapperClass(MyMapper.class);
    		//map输出的<k,v>类型。如果<k3,v3>的类型与<k2,v2>的类型一致，则可以省略
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(LongWritable.class);
    		
    		//1.3分区
    //		job.setPartitionerClass(HashPartitioner.class);
    		//有一个reudce任务运行
    //		job.setNumReduceTasks(1);
    		
    		//1.4 排序、分组
    		
    		
    		//1.5规约
    //		job.setCombinerClass(MyReduce.class);
    		
    		//2.2指定自定义的reduce类
    		job.setReducerClass(MyReduce.class);
    		//指定reduce的输出类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(LongWritable.class);
    		
    		//删除已存在的文件
    		FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), new Configuration());
    		Path path = new Path(OUT_PATH);
    		if(fileSystem.exists(path)){
    			fileSystem.delete(path, true);
    		}
    		
    		//2.3指定写出到哪里
    		FileOutputFormat.setOutputPath(job,new Path(OUT_PATH) );
    		
    		//指定输出文件的格式化类
    		job.setOutputFormatClass(TextOutputFormat.class);
    		
    		//把Job提交给JobTracker执行
    		job.waitForCompletion(true);
    		
    		
    	}
    	
    	/**
    	 * @ClassName: MyMapper 
    	 * @Description: map任务处理
    	 * @param  KEYIN 即k1  表示行的偏移量
    	 * @param  VALUEIN 即v1  表示行的文本内容
    	 * @param  KEYOUT 即k2  表示行中出现的单词
    	 * @param  VALUEOUT 即v2  表示行中出现的单词数，固定值为1
    	 * @author zhangweixiang
    	 * @date 2014年3月4日 下午4:16:00
    	 */
    	static class  MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
    		@Override
    		protected void map(LongWritable key, Text value,Context context)
    				throws IOException, InterruptedException {
    			String string = value.toString();
    			//自定义计数器(查询hello的出现，以及出现次数)
    			Counter counter = context.getCounter("查找hello", "hello");
    			if(string.contains("hello")){
    				counter.increment(1l);//出现一次+1
    			}
    			//分割字符串
    			String[] split = string.split(" ");
    			for(String word:split){
    				context.write(new Text(word), new LongWritable(1));//map任务输出
    			}
    		}
    	}
    	
    	/**
    	 * @ClassName: MyReduce 
    	 * @Description: reduce任务处理
    	 * @param KEYIN 即k2 表示行中出现的单词
    	 * @param VALUEIN 即v2 表示行中出现的单词个数
    	 * @param KEYOUT 即k3 表示文本中出现的不同单词
    	 * @param VALUEOUT 即v3 表示文本中出现不同单词的总次数
    	 * @author zhangweixiang
    	 * @date 2014年3月4日 下午4:23:20
    	 */
    	static class MyReduce extends Reducer<Text, LongWritable, Text, LongWritable>{
    		
    		protected void reduce(Text key, Iterable<LongWritable> values,Context context)
    				throws IOException, InterruptedException {
    				long sum=0;
    				for (LongWritable longWritable : values) {
    					sum+=longWritable.get();
    				}
    				context.write(key, new LongWritable(sum));//reduce输出
    		}
    	}
    }

code2

package combine;
    
    import java.io.IOException;
    import java.net.URI;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Counter;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    /**
     * @ClassName: WordCount2 
     * @Description: TODO(这里用一句话描述这个类的作用) 
     * @author zhangweixiang
     * @date 2014年3月6日 下午1:34:50
     */
    
    public class WordCount2 {
    	
    	static final String INPUT_PATH="hdfs://192.168.0.9:9000/hello.txt";
    	static final String OUT_PATH="hdfs://192.168.0.9:9000/word";
    	
    	public static void main(String[] args) throws Exception {
    		Configuration conf=new Configuration();
    		
    		Job job=new Job(conf, WordCount2.class.getSimpleName());
    		
    		//1.1指定读取的文件位置
    		FileInputFormat.addInputPaths(job, INPUT_PATH);
    		//指定如何对输入文件进行格式化，把输入文件每一行解析成键值对
    		job.setInputFormatClass(TextInputFormat.class);
    		
    		//1.2指定自定义的map类
    		job.setMapperClass(MyMapper.class);
    		//map输出的<k,v>类型。如果<k3,v3>的类型与<k2,v2>的类型一致，则可以省略
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(LongWritable.class);
    		
    		//1.3分区
    //		job.setPartitionerClass(HashPartitioner.class);
    		//有一个reudce任务运行
    //		job.setNumReduceTasks(1);
    		
    		//1.4 排序、分组
    		
    		
    		//1.5规约
    		job.setCombinerClass(MyReduce.class);
    		
    		//2.2指定自定义的reduce类
    		job.setReducerClass(MyReduce.class);
    		//指定reduce的输出类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(LongWritable.class);
    		
    		//删除已存在的文件
    		FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), new Configuration());
    		Path path = new Path(OUT_PATH);
    		if(fileSystem.exists(path)){
    			fileSystem.delete(path, true);
    		}
    		
    		//2.3指定写出到哪里
    		FileOutputFormat.setOutputPath(job,new Path(OUT_PATH) );
    		
    		//指定输出文件的格式化类
    		job.setOutputFormatClass(TextOutputFormat.class);
    		
    		//把Job提交给JobTracker执行
    		job.waitForCompletion(true);
    		
    		
    	}
    	
    	/**
    	 * @ClassName: MyMapper 
    	 * @Description: map任务处理
    	 * @param  KEYIN 即k1  表示行的偏移量
    	 * @param  VALUEIN 即v1  表示行的文本内容
    	 * @param  KEYOUT 即k2  表示行中出现的单词
    	 * @param  VALUEOUT 即v2  表示行中出现的单词数，固定值为1
    	 * @author zhangweixiang
    	 * @date 2014年3月4日 下午4:16:00
    	 */
    	static class  MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
    		@Override
    		protected void map(LongWritable key, Text value,Context context)
    				throws IOException, InterruptedException {
    			String string = value.toString();
    			//自定义计数器(查询hello的出现，以及出现次数)
    			Counter counter = context.getCounter("查找hello", "hello");
    			if(string.contains("hello")){
    				counter.increment(1l);//出现一次+1
    			}
    			//分割字符串
    			String[] split = string.split(" ");
    			for(String word:split){
    				context.write(new Text(word), new LongWritable(1));//map任务输出
    			}
    		}
    	}
    	
    	/**
    	 * @ClassName: MyReduce 
    	 * @Description: reduce任务处理
    	 * @param KEYIN 即k2 表示行中出现的单词
    	 * @param VALUEIN 即v2 表示行中出现的单词个数
    	 * @param KEYOUT 即k3 表示文本中出现的不同单词
    	 * @param VALUEOUT 即v3 表示文本中出现不同单词的总次数
    	 * @author zhangweixiang
    	 * @date 2014年3月4日 下午4:23:20
    	 */
    	static class MyReduce extends Reducer<Text, LongWritable, Text, LongWritable>{
    		
    		protected void reduce(Text key, Iterable<LongWritable> values,Context context)
    				throws IOException, InterruptedException {
    				long sum=0;
    				for (LongWritable longWritable : values) {
    					sum+=longWritable.get();
    				}
    				context.write(key, new LongWritable(sum));//reduce输出
    		}
    	}
    }

code1和code2的唯一区别就是  
//1.5规约  
job.setCombinerClass(MyReduce.class);  
code1中加了注释，code2启用了规约！  
接下来看看控制台的输出（我win8下的eclipse）  
code1:  
/06 13:52:17 INFO mapred.JobClient: Counters: 20  
14/03/06 13:52:17 INFO mapred.JobClient:   File Output Format Counters   
14/03/06 13:52:17 INFO mapred.JobClient:     Bytes Written=28  
14/03/06 13:52:17 INFO mapred.JobClient:   FileSystemCounters  
14/03/06 13:52:17 INFO mapred.JobClient:     FILE\_BYTES\_READ=424  
14/03/06 13:52:17 INFO mapred.JobClient:     HDFS\_BYTES\_READ=84  
14/03/06 13:52:17 INFO mapred.JobClient:     FILE\_BYTES\_WRITTEN=128676  
14/03/06 13:52:17 INFO mapred.JobClient:     HDFS\_BYTES\_WRITTEN=28  
14/03/06 13:52:17 INFO mapred.JobClient:   File Input Format Counters   
14/03/06 13:52:17 INFO mapred.JobClient:     Bytes Read=42  
14/03/06 13:52:17 INFO mapred.JobClient:   查找hello  
14/03/06 13:52:17 INFO mapred.JobClient:     hello=2  
14/03/06 13:52:17 INFO mapred.JobClient:   Map-Reduce Framework  
14/03/06 13:52:17 INFO mapred.JobClient:     Map output materialized bytes=126  
14/03/06 13:52:17 INFO mapred.JobClient:     Map input records=4  
14/03/06 13:52:17 INFO mapred.JobClient:     Reduce shuffle bytes=0  
14/03/06 13:52:17 INFO mapred.JobClient:     Spilled Records=16  
14/03/06 13:52:17 INFO mapred.JobClient:     Map output bytes=104  
14/03/06 13:52:17 INFO mapred.JobClient:     Total committed heap usage (bytes)=266469376  
14/03/06 13:52:17 INFO mapred.JobClient:     SPLIT\_RAW\_BYTES=98  
14/03/06 13:52:17 INFO mapred.JobClient:     Combine input records=0  
14/03/06 13:52:17 INFO mapred.JobClient:     Reduce input records=8  
14/03/06 13:52:17 INFO mapred.JobClient:     Reduce input groups=4  
14/03/06 13:52:17 INFO mapred.JobClient:     Combine output records=0  
14/03/06 13:52:17 INFO mapred.JobClient:     Reduce output records=4  
14/03/06 13:52:17 INFO mapred.JobClient:     Map output records=8  
  
  
code2:  
14/03/06 13:55:11 INFO mapred.JobClient: Counters: 20  
14/03/06 13:55:11 INFO mapred.JobClient:   File Output Format Counters   
14/03/06 13:55:11 INFO mapred.JobClient:     Bytes Written=28  
14/03/06 13:55:11 INFO mapred.JobClient:   FileSystemCounters  
14/03/06 13:55:11 INFO mapred.JobClient:     FILE\_BYTES\_READ=364  
14/03/06 13:55:11 INFO mapred.JobClient:     HDFS\_BYTES\_READ=84  
14/03/06 13:55:11 INFO mapred.JobClient:     FILE\_BYTES\_WRITTEN=129072  
14/03/06 13:55:11 INFO mapred.JobClient:     HDFS\_BYTES\_WRITTEN=28  
14/03/06 13:55:11 INFO mapred.JobClient:   File Input Format Counters   
14/03/06 13:55:11 INFO mapred.JobClient:     Bytes Read=42  
14/03/06 13:55:11 INFO mapred.JobClient:   查找hello  
14/03/06 13:55:11 INFO mapred.JobClient:     hello=2  
14/03/06 13:55:11 INFO mapred.JobClient:   Map-Reduce Framework  
14/03/06 13:55:11 INFO mapred.JobClient:     Map output materialized bytes=66  
14/03/06 13:55:11 INFO mapred.JobClient:     Map input records=4  
14/03/06 13:55:11 INFO mapred.JobClient:     Reduce shuffle bytes=0  
14/03/06 13:55:11 INFO mapred.JobClient:     Spilled Records=8  
14/03/06 13:55:11 INFO mapred.JobClient:     Map output bytes=104  
14/03/06 13:55:11 INFO mapred.JobClient:     Total committed heap usage (bytes)=266469376  
14/03/06 13:55:11 INFO mapred.JobClient:     SPLIT\_RAW\_BYTES=98  
14/03/06 13:55:11 INFO mapred.JobClient:     Combine input records=8  
14/03/06 13:55:11 INFO mapred.JobClient:     Reduce input records=4  
14/03/06 13:55:11 INFO mapred.JobClient:     Reduce input groups=4  
14/03/06 13:55:11 INFO mapred.JobClient:     Combine output records=4  
14/03/06 13:55:11 INFO mapred.JobClient:     Reduce output records=4  
14/03/06 13:55:11 INFO mapred.JobClient:     Map output records=8  
  
  
通过code1和code2的输出比较（红色部分），相信大家已经看出了区别！  
code1中没有使用规约，所以：  
Combine input records=0  
Reduce input records=8  
 Reduce input groups=4  
Combine output records=0  
code2中使用了规约，所以：  
 Combine input records=8  
 Reduce input records=4  
Reduce input groups=4  
Combine output records=4  
  
  
很明显使用了规约后 Reduce input records=4减少map到reduce过程中数据的传输！  
  
  
为什么要使用规约呢？（总结超哥的！）  
/\*\*  
 \* 问：为什么使用Combiner？  
 \* 答：Combiner发生在Map端，对数据进行规约处理，数据量变小了，传送到reduce端的数据量变小了，传输时间变短，作业的整体时间变短。  
 \*   
 \* 问：为什么Combiner不作为MR运行的标配，而是可选步骤哪？  
 \* 答：因为不是所有的算法都适合使用Combiner处理，例如求平均数。  
 \*  
 \* 问：Combiner本身已经执行了reduce操作，为什么在Reducer阶段还要执行reduce操作哪？  
 \* 答：combiner操作发生在map端的，处理一个任务所接收的文件中的数据，不能跨map任务执行；只有reduce可以接收多个map任务处理的数据。  
 \*  
 \*/