MapReduce在Map端的Combiner和在Reduce端的Partitioner

 1 /**
 2  * 以文本
 3  * hello    you
 4  * hello    me
 5  * 为例子.
 6  * map方法调用了两次,因为有两行
 7  * k2 v2 键值对的数量有几个?
 8  * 有4个.有四个单词.
 9  * 
10  * 会产生几个分组?
11  * 产生3个分组.
12  * 有3个不同的单词.
13  *
14  */
15 public class WordCountApp {
16     public static void main(String[] args) throws Exception {
17         //程序在这里运行,要有驱动.
18         Configuration conf = new Configuration();
19         Job job = Job.getInstance(conf,WordCountApp.class.getSimpleName());
20         
21         //我们运行此程序通过运行jar包来执行.一定要有这句话.
22         job.setJarByClass(WordCountApp.class);
23         FileInputFormat.setInputPaths(job,args[0]);
24         
25         job.setMapperClass(WordCountMapper.class);//设置Map类
26         job.setMapOutputKeyClass(Text.class);//设置Map的key
27         job.setMapOutputValueClass(LongWritable.class);//设置Map的value
28         
29         job.setCombinerClass(WordCountReducer.class);//数据在Map端先进行 一次合并.
30         /*
31              这个setCombinerClass设置参数只能是一个继承了Reduce类的类.直接用我们定义的WordCountReducer.
32             在单词技术的例子中,Map端产生了四个键值对,两个hello,you和me各一个.
33             这样合并之后Map端最终只产生三个键值对.
34             这样在Reduce端也只处理三个键值对,而不是没有合并之前的四个.
35             这样Map端最终产生的键值对少了,Map端向Reduce端传递键值对占用的带宽就小.提高网络通信的速度.
36             Reduce端接受键值对的数量变少,就减少了Reduce端处理键值对所需要的时间.
37             以上就是Combiner的好处(在Map端对数据进行一次合并).
38             Map端的合并和Reduce端的合并是不能相互取代的.
39             在Map端进行的合并是局部合并,当前Map任务在它之中的合并.
40             各个Map任务之间还是会 有相同的数据的.这些相同的数据要到Reduce端进行合并.
41          */
42         
43         job.setReducerClass(WordCountReducer.class);//设置Reduce的类
44         job.setOutputKeyClass(Text.class);//设置Reduce的key Reduce这个地方只有输出的参数可以设置. 方法名字也没有Reduce关键字区别于Map
45         job.setOutputValueClass(LongWritable.class);//设置Reduce的value.
46         
47         FileOutputFormat.setOutputPath(job, new Path(args[1]));
48         job.waitForCompletion(true);//表示结束了才退出,不结束不退出
49     }

......................................................

 1 public class TrafficApp {
 2     public static void main(String[] args) throws Exception {
 3         Job job = Job.getInstance(new Configuration(), TrafficApp.class.getSimpleName());
 4         job.setJarByClass(TrafficApp.class);
 5         
 6         FileInputFormat.setInputPaths(job, args[0]);
 7         
 8         job.setMapperClass(TrafficMapper.class);
 9         job.setMapOutputKeyClass(Text.class);
10         job.setMapOutputValueClass(TrafficWritable.class);
11         
12         job.setNumReduceTasks(2);//设定Reduce的数量为2
13         job.setPartitionerClass(TrafficPartitioner.class);//设定一个Partitioner的类.
14         /*
15          *Partitioner是如何实现不同的Map输出分配到不同的Reduce中?
16          *在不适用指定的Partitioner时,有 一个默认的Partitioner.
17          *就是HashPartitioner. 
18          *其只有一行代码,其意思就是过来的key,不管是什么,模numberReduceTasks之后 返回值就是reduce任务的编号.
19          *numberReduceTasks的默认值是1.  任何一个数模1(取余数)都是0. 
20          *这个地方0就是取编号为0的Reduce.(Reduce从0开始编号.) 
21          */
22         
23         job.setReducerClass(TrafficReducer.class);
24         job.setOutputKeyClass(Text.class);
25         job.setOutputValueClass(TrafficWritable.class);
26         
27         FileOutputFormat.setOutputPath(job, new Path(args[1]));
28         job.waitForCompletion(true);
29     }
30     
31     public static class TrafficPartitioner extends Partitioner<Text,TrafficWritable>{//k2,v2
32 
33         @Override
34         public int getPartition(Text key, TrafficWritable value,int numPartitions) {
35             long phoneNumber = Long.parseLong(key.toString());
36             return (int)(phoneNumber%numPartitions);
37         }
38         
39     }

.................................................