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

1.Map端的Combiner.

通过单词计数WordCountApp.java的例子,如何在Map端设置Combiner...

只附录部分代码:

/**
* 以文本
* hello    you
* hello    me
* 为例子.
* map方法调用了两次,因为有两行
* k2 v2 键值对的数量有几个?
* 有4个.有四个单词.
*
* 会产生几个分组?
* 产生3个分组.
* 有3个不同的单词.
*
*/
public class WordCountApp {
public static void main(String[] args) throws Exception {
//程序在这里运行,要有驱动.
Configuration conf = new Configuration();
Job job = Job.getInstance(conf,WordCountApp.class.getSimpleName());

//我们运行此程序通过运行jar包来执行.一定要有这句话.
job.setJarByClass(WordCountApp.class);
FileInputFormat.setInputPaths(job,args[0]);

job.setMapperClass(WordCountMapper.class);//设置Map类
job.setMapOutputKeyClass(Text.class);//设置Map的key
job.setMapOutputValueClass(LongWritable.class);//设置Map的value

job.setCombinerClass(WordCountReducer.class);//数据在Map端先进行 一次合并.
/*
这个setCombinerClass设置参数只能是一个继承了Reduce类的类.直接用我们定义的WordCountReducer.
在单词技术的例子中,Map端产生了四个键值对,两个hello,you和me各一个.
这样合并之后Map端最终只产生三个键值对.
这样在Reduce端也只处理三个键值对,而不是没有合并之前的四个.
这样Map端最终产生的键值对少了,Map端向Reduce端传递键值对占用的带宽就小.提高网络通信的速度.
Reduce端接受键值对的数量变少,就减少了Reduce端处理键值对所需要的时间.
以上就是Combiner的好处(在Map端对数据进行一次合并).
Map端的合并和Reduce端的合并是不能相互取代的.
在Map端进行的合并是局部合并,当前Map任务在它之中的合并.
各个Map任务之间还是会 有相同的数据的.这些相同的数据要到Reduce端进行合并.
*/

job.setReducerClass(WordCountReducer.class);//设置Reduce的类
job.setOutputKeyClass(Text.class);//设置Reduce的key Reduce这个地方只有输出的参数可以设置. 方法名字也没有Reduce关键字区别于Map
job.setOutputValueClass(LongWritable.class);//设置Reduce的value.

FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.waitForCompletion(true);//表示结束了才退出,不结束不退出
}
......................................................

2.Reduce端的Partitioner.

以流量统计TrafficCountApp.java的例子示例Reduce端设置Partitioner.

只附录部分代码:

public class TrafficApp {
public static void main(String[] args) throws Exception {
Job job = Job.getInstance(new Configuration(), TrafficApp.class.getSimpleName());
job.setJarByClass(TrafficApp.class);

FileInputFormat.setInputPaths(job, args[0]);

job.setMapperClass(TrafficMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(TrafficWritable.class);

job.setNumReduceTasks(2);//设定Reduce的数量为2
job.setPartitionerClass(TrafficPartitioner.class);//设定一个Partitioner的类.
/*
*Partitioner是如何实现不同的Map输出分配到不同的Reduce中?
*在不适用指定的Partitioner时,有 一个默认的Partitioner.
*就是HashPartitioner.
*其只有一行代码,其意思就是过来的key,不管是什么,模numberReduceTasks之后 返回值就是reduce任务的编号.
*numberReduceTasks的默认值是1.  任何一个数模1(取余数)都是0.
*这个地方0就是取编号为0的Reduce.(Reduce从0开始编号.)
*/

job.setReducerClass(TrafficReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(TrafficWritable.class);

FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.waitForCompletion(true);
}

public static class TrafficPartitioner extends Partitioner<Text,TrafficWritable>{//k2,v2

@Override
public int getPartition(Text key, TrafficWritable value,int numPartitions) {
long phoneNumber = Long.parseLong(key.toString());
return (int)(phoneNumber%numPartitions);
}

}
.................................................

//============附录MapReduce中Reduce使用默认的HashPartitioner进行分组的源代码==============