MapReduce设计模式--分区

     分区模式将记录进行分类，但它并不关心记录的顺序。其主要目的是将数据集中相似的记录分成不同的、更小的数据集
     分区主要原理是，自定义分区类继承Partitioner，根据业务需求实现分区函数 public int getPartition(Text key, Text value, int numPartitions)，将Key相同的记录，划分到同一reduce函数中。需要注意的是如果在驱动程序中将NumReduceTasks值设置为时， 不会执行分区函数。这个可以理解，毕竟只有1个reduce，所以，没有必要执行Partitioner
 
具体实现如下：

点击(此处)折叠或打开

1. import org.apache.hadoop.fs.FileSystem;
   
2. import org.apache.hadoop.fs.Path;
   
3. import org.apache.hadoop.io.LongWritable;
   
4. import org.apache.hadoop.io.NullWritable;
   
5. import org.apache.hadoop.io.Text;
   
6. import org.apache.hadoop.mapreduce.Job;
   
7. import org.apache.hadoop.mapreduce.Mapper;
   
8. import org.apache.hadoop.mapreduce.Partitioner;
   
9. import org.apache.hadoop.mapreduce.Reducer;
   
10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    
11. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
12. 
    
13. import java.io.IOException;
    
14. 
    
15. public class PartitionJob {
    
16. 
    
17.     public static class PartitionMapper extends Mapper<LongWritable ,Text, Text,Text>{
    
18.         private Text tmpKey = new Text();
    
19.         @Override
    
20.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    
21.             String data = value.toString();
    
22.             String []arrays = data.split("\t");
    
23.             tmpKey.set(arrays[0]);
    
24.             System.out.println("data:"+data);
    
25.             context.write(tmpKey, value);
    
26.         }
    
27.     }
    
28. 
    
29.     public static class PartitionReduce extends Reducer<Text, Text, NullWritable, Text>{
    
30.         private Text out = new Text();
    
31.         @Override
    
32.         protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    
33.             String str="";
    
34.             for (Text val:values){
    
35.                 context.write(NullWritable.get(), val);
    
36.             }
    
37.         }
    
38. 
    
39.     }
    
40. 
    
41.     public static class Partition extends Partitioner<Text ,Text>{
    
42.         @Override
    
43.         public int getPartition(Text key, Text value, int numPartitions) {
    
44.             return Math.abs(key.hashCode())%numPartitions;
    
45.         }
    
46. 
    
47.     }
    
48. 
    
49.     public static void main(String []args){
    
50. 
    
51.         try {
    
52.             Job job = Job.getInstance();
    
53.             job.setJobName("PartitionJob");
    
54.             job.setJarByClass(PartitionJob.class);
    
55.             job.setMapperClass(PartitionMapper.class);
    
56.             job.setMapOutputKeyClass(Text.class);
    
57.             job.setMapOutputValueClass(Text.class);
    
58. 
    
59.             job.setReducerClass(PartitionReduce.class);
    
60.             job.setOutputKeyClass(NullWritable.class);
    
61.             job.setOutputValueClass(Text.class);
    
62.             job.setPartitionerClass(Partition.class);
    
63.             job.setNumReduceTasks(2);
    
64. 
    
65.             FileInputFormat.addInputPath(job, new Path(args[0]));
    
66.             FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
67.             FileSystem.get(job.getConfiguration()).delete(new Path(args[1]), true);
    
68. 
    
69.            System.out.println(job.waitForCompletion(true));
    
70.         } catch (IOException e) {
    
71.             e.printStackTrace();
    
72.         } catch (InterruptedException e) {
    
73.             e.printStackTrace();
    
74.         } catch (ClassNotFoundException e) {
    
75.             e.printStackTrace();
    
76.         }
    
77. 
    
78. 
    
79.     }
    
80. }