MapReduce中的分区默认是哈希分区,根据map输出key的哈希值做模运算,如下
int result = key.hashCode()%numReduceTask;
如果我们需要根据业务需求来将map读入的数据按照某些特定条件写入不同的文件,那就需要自定义实现Partition,自定义规则
举个简单的例子,使用MapReduce做wordcount,但是需要根据单词的长度写入不同的文件中,单词的长度大于4的写入一个文件,小于等于4的写入另一个文件
代码结构如下
代码实现如下
MapTest.java
/** * */ package com.zhen.partition; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; /** * @author FengZhen * */ public class MapTest extends Mapper<LongWritable, Text, Text, IntWritable>{ private IntWritable outputValue = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException { String[] splits = value.toString().split(" "); for (int i = 0; i < splits.length; i++) { context.write(new Text(splits[i]), outputValue); } } }
ReduceTest.java
/** * */ package com.zhen.partition; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; /** * @author FengZhen * */ public class ReduceTest extends Reducer<Text, IntWritable, Text, IntWritable>{ @Override protected void reduce(Text key, Iterable<IntWritable> value, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable intWritable : value) { sum += intWritable.get(); } context.write(key, new IntWritable(sum)); } }
PartitionTest.java
/** * */ package com.zhen.partition; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; /** * @author FengZhen * 第一个参数:map的输出key类型 * 第二个参数:map的输出value类型 */ public class PartitionTest extends Partitioner<Text, IntWritable>{ /** * key:map的输出key * value:mapd的输出value * numReduceTask:reduce的task数量 * 返回值,指定reduce,从0开始 * */ @Override public int getPartition(Text key, IntWritable value, int numReduceTask) { if (key.toString().length()>4) { return 0; }else{ return 1; } } }
PartitionTestMain.java
/** * */ package com.zhen.partition; import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; 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.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * @author FengZhen * */ public class PartitionTestMain { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration configuration = new Configuration(); Job job = new Job(configuration, PartitionTestMain.class.getSimpleName()); job.setJarByClass(PartitionTestMain.class); job.setMapperClass(MapTest.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setReducerClass(ReduceTest.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setCombinerClass(ReduceTest.class);
//设置分区类 job.setPartitionerClass(PartitionTest.class);
//设置reduce任务个数 job.setNumReduceTasks(2); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true)?0:1); } }
打包测试
hadoop jar /Users/FengZhen/Desktop/Hadoop/other/mapreduce_jar/PartitionTest.jar com.zhen.partition.PartitionTestMain /user/hadoop/mapreduce/partitionTest/input /user/hadoop/mapreduce/partitionTest/output/
任务结束后可看到输出路径下有两个结果文件
EFdeMacBook-Pro:file FengZhen$ hadoop fs -ls /user/hadoop/mapreduce/partitionTest/output/ Found 3 items -rw-r--r-- 1 FengZhen supergroup 0 2018-02-11 12:12 /user/hadoop/mapreduce/partitionTest/output/_SUCCESS -rw-r--r-- 1 FengZhen supergroup 69 2018-02-11 12:12 /user/hadoop/mapreduce/partitionTest/output/part-r-00000 -rw-r--r-- 1 FengZhen supergroup 19 2018-02-11 12:12 /user/hadoop/mapreduce/partitionTest/output/part-r-00001
查看文件内容,是按照条件来分别输出的
part-r-00000中是length > 4的单词
part-r-00001中是length <= 4的单词