12.hadoop系列之MapReduce分区实践

本文我们学习MapReduce默认分区以及自定义分区实践

当我们要求将统计结果按照条件输出到不同文件(分区)，比如按照统计结果将手机归属地不同省份输出到不同文件中(分区)

1.默认Partitioner分区

public class HashPartitioner<K, V> extends Partitioner<K, V> {public int getPartition(K key, V value,int numReduceTasks) {return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;}
}

默认分区是根据key的hashCode对ReduceTasks[通过job.setNumReduceTasks(2)赋值]取模得到，用户没法控制key存储到哪个分区

2. 自定义Partitioner分区

• 我们在resources目录下新建phone2.txt

1 13764368888 196.168.0.11 1116 854 200
2 13764368888 196.168.0.11 1136 834 200
3 13764368888 196.168.0.11 1146 824 200
4 13764368888 196.168.0.11 1116 804 200
5 13664368888 196.168.0.11 1116 854 200
6 13864368888 196.168.0.11 1136 834 200
7 13964368888 196.168.0.11 1146 824 200
8 13764368888 196.168.0.11 1116 804 200

• 新建自定义ProvincePartitioner类

public class ProvincePartitioner extends Partitioner<Text, FlowBean> {@Overridepublic int getPartition(Text text, FlowBean flowBean, int numPartitions) {// Text是手机号String phone = text.toString().substring(0, 3);// 注意分区号需要连续，从0开始分区int partition;if ("136".equals(phone)) {partition = 0;} else if ("137".equals(phone)) {partition = 1;} else if ("138".equals(phone)) {partition = 2;} else if ("139".equals(phone)) {partition = 3;} else {partition = 4;}return partition;}
}

• 新建FlowPartitionerDriver类

public class FlowPartitionerDriver {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {Configuration conf = new Configuration();Job job = Job.getInstance(conf, "flowPartitioner");job.setJarByClass(FlowPartitionerDriver.class);job.setMapperClass(FlowMapper.class);job.setCombinerClass(FlowReduce.class);job.setReducerClass(FlowReduce.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);// 关联自定义分区类job.setPartitionerClass(ProvincePartitioner.class);// 设置ReduceTask任务数job.setNumReduceTasks(5);FileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));System.exit(job.waitForCompletion(true) ? 0 : 1);}
}

• 传参运行

E:\Java\blogCode\hadoop\src\main\resources\phone2.txt E:\Java\blogCode\hadoop\src\main\resources\phone_ret2.txt

由图可知，产生了5个分区，实现了手机归属地不同省份输出到不同文件中

3.分区总结

1. 如果ReduceTask数量>getPartition结果数，则会多产生空的part-r-000xx文件
2. 如果1<ReduceTask数量<getPartition结果数,则有一部分数据无处写，会Exception
3. 如果ReduceTask数量=1，则不管MapTask输出多少分区文件，最终结果都会交给一个ReduceTask,只会产生一个文件part-r-00000
4. 分区号必须从零开始，逐一累加

欢迎关注公众号算法小生与我沟通交流