MapReduce之自定义分区器Partitioner

@

目录

• 问题引出
• 默认Partitioner分区
• 自定义Partitioner步骤
• Partition分区案例实操
• 分区总结

问题引出

要求将统计结果按照条件输出到不同文件中（分区）。

比如：将统计结果按照手机归属地不同省份输出到不同文件中（分区）

默认Partitioner分区

1public class HashPartitioner<K,V> extends Partitioner<K,V>{
2	public int getPartition(K key,V value, int numReduceTasks){
3		return (key.hashCode() & Integer.MAX VALUE) & numReduceTasks;
4	}
5}

• 默认分区是根据key的hashCode对ReduceTasks个数取模得到的。
• 用户没法控制哪个key存储到哪个分区。

自定义Partitioner步骤

1. 自定义类继承Partitioner，重写getPartition()方法

   public class CustomPartitioner extends Partitioner<Text，FlowBea>{ @Override public int getPartition(Text key，FlowBean value，int numPartitions){ //控制分区代码逻辑 …… return partition； } }

2. 在Job驱动类中，设置自定义Partitioner

   job.setPartitionerClass(CustomPartitioner.class)

3. 自定义Partition后，要根据自定义Partitioner的逻辑设置相应数量的ReduceTask

   job.setNumReduceTask(5);//假设需要分5个区

Partition分区案例实操

将统计结果按照手机归属地不同省份输出到不同文件中（分区）

输入数据：

期望输出数据：
手机号136、137、138、139开头都分别放到一个独立的4个文件中，其他开头的放到一个文件中。所以总共分为5个文件，也就是五个区。

相比于之前的自定义flowbean，这次自定义分区，只需要多编写一个分区器，以及在job驱动类中设置分区器，mapper和reducer类不改变

MyPartitioner.java

1/*
2 * KEY, VALUE: Mapper输出的Key-value类型
3 */
4public class MyPartitioner extends Partitioner<Text, FlowBean>{
5
6	// 计算分区  numPartitions为总的分区数，reduceTask的数量
7	// 分区号必须为int型的值，且必须符合 0<= partitionNum < numPartitions
8	@Override
9	public int getPartition(Text key, FlowBean value, int numPartitions) {
10		
11		String suffix = key.toString().substring(0, 3);//前开后闭，取手机号前三位数
12		
13		int partitionNum=0;//分区编号
14		
15		
16		switch (suffix) {
17		case "136":
18			partitionNum=numPartitions-1;//由于分区编号不能大于分区总数，所以用这种方法比较好
19			break;
20		case "137":
21			partitionNum=numPartitions-2;
22			break;
23		case "138":
24			partitionNum=numPartitions-3;
25			break;
26		case "139":
27			partitionNum=numPartitions-4;
28			break;
29
30		default:
31			break;
32		}
33
34		return partitionNum;
35	}
36
37}

FlowBeanDriver.java

1public class FlowBeanDriver {
2	
3	public static void main(String[] args) throws Exception {
4		
5		Path inputPath=new Path("e:/mrinput/flowbean");
6		Path outputPath=new Path("e:/mroutput/partitionflowbean");
7		
8		//作为整个Job的配置
9		Configuration conf = new Configuration();
10		
11		//保证输出目录不存在
12		FileSystem fs=FileSystem.get(conf);
13		
14		if (fs.exists(outputPath)) {
15			fs.delete(outputPath, true);
16		}
17		
18		// ①创建Job
19		Job job = Job.getInstance(conf);
20		
21		// ②设置Job
22		// 设置Job运行的Mapper，Reducer类型，Mapper,Reducer输出的key-value类型
23		job.setMapperClass(FlowBeanMapper.class);
24		job.setReducerClass(FlowBeanReducer.class);
25		
26		// Job需要根据Mapper和Reducer输出的Key-value类型准备序列化器，通过序列化器对输出的key-value进行序列化和反序列化
27		// 如果Mapper和Reducer输出的Key-value类型一致，直接设置Job最终的输出类型
28		job.setOutputKeyClass(Text.class);
29		job.setOutputValueClass(FlowBean.class);
30		
31		// 设置输入目录和输出目录
32		FileInputFormat.setInputPaths(job, inputPath);
33		FileOutputFormat.setOutputPath(job, outputPath);
34		
35		// 设置ReduceTask的数量为5
36		job.setNumReduceTasks(5);
37		
38		// 设置使用自定义的分区器
39		job.setPartitionerClass(MyPartitioner.class);
40		
41		// ③运行Job
42		job.waitForCompletion(true);
43		
44	}
45}

FlowBeanMapper.java

1/*
2 * 1. 统计手机号(String)的上行(long,int)，下行(long,int)，总流量(long,int)
3 * 
4 * 手机号为key,Bean{上行(long,int)，下行(long,int)，总流量(long,int)}为value
5 * 		
6 * 
7 * 
8 * 
9 */
10public class FlowBeanMapper extends Mapper<LongWritable, Text, Text, FlowBean>{
11	
12	private Text out_key=new Text();
13	private FlowBean out_value=new FlowBean();
14	
15	// (0,1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200)
16	@Override
17	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context)
18			throws IOException, InterruptedException {
19		
20		String[] words = value.toString().split("\t");
21		
22		//封装手机号
23		out_key.set(words[1]);
24		// 封装上行
25		out_value.setUpFlow(Long.parseLong(words[words.length-3]));
26		// 封装下行
27		out_value.setDownFlow(Long.parseLong(words[words.length-2]));
28
29		context.write(out_key, out_value);
30	}
31}

FlowBeanReducer.java

1public class FlowBeanReducer extends Reducer<Text, FlowBean, Text, FlowBean>{
2	
3	private FlowBean out_value=new FlowBean();
4	
5	@Override
6	protected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context)
7			throws IOException, InterruptedException {
8		
9		long sumUpFlow=0;
10		long sumDownFlow=0;
11		
12		for (FlowBean flowBean : values) {
13			
14			sumUpFlow+=flowBean.getUpFlow();
15			sumDownFlow+=flowBean.getDownFlow();
16			
17		}
18		
19		out_value.setUpFlow(sumUpFlow);
20		out_value.setDownFlow(sumDownFlow);
21		out_value.setSumFlow(sumDownFlow+sumUpFlow);
22		
23		context.write(key, out_value);
24		
25	}
26}

FlowBean.java

1public class FlowBean implements Writable{
2	
3	private long upFlow;
4	private long downFlow;
5	private long sumFlow;
6	
7	public FlowBean() {
8		
9	}
10
11	public long getUpFlow() {
12		return upFlow;
13	}
14
15	public void setUpFlow(long upFlow) {
16		this.upFlow = upFlow;
17	}
18
19	public long getDownFlow() {
20		return downFlow;
21	}
22
23	public void setDownFlow(long downFlow) {
24		this.downFlow = downFlow;
25	}
26
27	public long getSumFlow() {
28		return sumFlow;
29	}
30
31	public void setSumFlow(long sumFlow) {
32		this.sumFlow = sumFlow;
33	}
34
35	// 序列化   在写出属性时，如果为引用数据类型，属性不能为null
36	@Override
37	public void write(DataOutput out) throws IOException {
38		
39		out.writeLong(upFlow);
40		out.writeLong(downFlow);
41		out.writeLong(sumFlow);
42		
43		
44	}
45
46	//反序列化   序列化和反序列化的顺序要一致
47	@Override
48	public void readFields(DataInput in) throws IOException {
49		upFlow=in.readLong();
50		downFlow=in.readLong();
51		sumFlow=in.readLong();
52		
53	}
54
55	@Override
56	public String toString() {
57		return  upFlow + "\t" + downFlow + "\t" + sumFlow;
58	}
59}

输出结果：
总共五个文件

一号区：

二号区：

三号区：

四号区：

其他号码为第五号区：

分区总结

• 如果ReduceTask的数量 > getPartition的结果数，则会多产生几个空的输出文件part-r-000xx
• 如果Reduceask的数量 < getPartition的结果数，则有一部分分区数据无处安放，会Exception
• 如果ReduceTask的数量 = 1，则不管MapTask端输出多少个分区文件，最终结果都交给这一个ReduceTask，最终也就只会产生一个结果文件partr-00000

以刚才的案例分析：
例如：假设自定义分区数为5，则

• job.setlNlurmReduce Task(1)；会正常运行，只不过会产生一个输出文件
• job.setlNlunReduce Task(2)，会报错
• job.setNumReduceTasks(6)；大于5，程序会正常运行，会产生空文件