本篇博客小菌为大家带来的是MapReduce的自定义分区与ReduceTask内容的分享(ReduceMap具体计算流程见《MapReduce中shuffle阶段概述及计算任务流程》)。
在MapReduce中。通过指定分区，会将同一个分区的数据发送到同一个reduce中，例如为了数据的统计，可以把一批类似的数据发 送到同一个reduce当中去，在同一个reduce中统计相同类型的数据，就可以实现类似数据的分区，统计等。 直观的说就是相同类型的数据，送到一起去处理，在reduce当中默认分区只有1个。
MapReduce当中的分区类图
假设现在有个业务需求,有一个文本文件partition.txt。其中第六个字段表示开奖结果数值，现在以15为分界点，将15以上的结果保存到一个文件，15以下的结果保存到一个文件。
那我们应该怎么做呢? 按照正常的思维,我们应该先创建两个目录,在Map中拿到数据后,把开奖结果数值作为key,整行数据作为value传递给reduce,然后reduce再根据key值大小做出一个判断,把list数据存入到不同的目录中…
虽然步骤没有错,但总归有些繁琐。事实上,MapReduce中的分区机制就可以很好的解决这个问题!
先让我们来看下MapReduce自带的默认分区算法:
对key 进行哈希，获取到一个哈希值，用这个哈希值与reducetask的数量取余。余几，这个数据就放在余数编号的partition中。
但很明显上面提到的问题最终是根据数值大小进行分区,所以这个分区算法并不适用,所以我们需要自定义分区!
自定义分区的步骤见下:
第一步：定义mapper
这里的mapper程序不做任何逻辑，也不对key，与value做任何改变，只是接收数据，然后往下发送。
代码语言：javascript复制import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class MyMapper extends Mapper<LongWritable,Text,Text, NullWritable>{
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
context.write(value,NullWritable.get());
}
}第二步：定义reducer逻辑
reducer也不做任何处理，将数据原封不动的输出即可。
代码语言：javascript复制import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class MyReducer extends Reducer<Text, NullWritable,Text,NullWritable> {
@Override
protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
context.write(key,NullWritable.get());
}
}第三步：自定义partitioner(分区)
代码语言：javascript复制import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
/**
* 这里的输入类型与map阶段的输出类型相同
*/
public class MyPartitioner extends Partitioner<Text,NullWritable> {
/**
* 返回值表示数据要去到哪个分区
* 返回值只是一个分区的标记，标记所有相同的数据去到指定的分区
*/
@Override
public int getPartition(Text text, NullWritable nullWritable, int i) {
String result = text.toString().split("t")[5];
System.out.println(result);
if (Integer.parseInt(result) > 15){
return 1;
}else{
return 0;
}
}
}第四步：程序main函数入口
代码语言：javascript复制import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class PartitionMain extends Configured implements Tool {
public static void main(String[] args) throws Exception{
int run = ToolRunner.run(new Configuration(), new PartitionMain(), args);
System.exit(run);
}
@Override
public int run(String[] args) throws Exception {
Job job = Job.getInstance(super.getConf(), PartitionMain.class.getSimpleName());
job.setJarByClass(PartitionMain.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
TextInputFormat.addInputPath(job,new Path("hdfs://192.168.100.129:8020/partitioner"));
TextOutputFormat.setOutputPath(job,new Path("hdfs://192.168.100.129:8020/outpartition"));
job.setMapperClass(MyMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(NullWritable.class);
job.setOutputKeyClass(Text.class);
job.setMapOutputValueClass(NullWritable.class);
job.setReducerClass(MyReducer.class);
/**
* 设置分区类，以及reducetask的个数，注意reduceTask的个数一定要与
* 分区数保持一致
*/
//使用combine
job.setPartitionerClass(MyPartitioner.class);
// 设置reduceTask的个数
job.setNumReduceTasks(2);
boolean b = job.waitForCompletion(true);
return b?0:1;
}
}注意：分区的案例，只能打成jar包发布到集群上面去运行，本地模式已经不能正常运行了
运行结果
可以发现,数值大于15的放在了一个分区,小于等于15的又放在了另一个分区,说明我们自定义分区成功了!
在这个过程中最最重要的一点小菌有必要再重复一遍,在设置分区类，以及reducetask的个数，注意reduceTask的个数一定要与分区数保持一致,否则分区将不具有任何意义!
本次的分享就到这里了,受益的小伙伴们记得点赞关注小菌~小菌后续会更新更多hadoop的相关内容,敬请关注(^_−)☆
