关于MapReduce中自定义分区类（四）

MapTask类

在MapTask类中找到run函数

1. if(useNewApi){
2.       runNewMapper(job, splitMetaInfo, umbilical, reporter);
3.     }

再找到runNewMapper

1. @SuppressWarnings("unchecked")
2.   private<INKEY,INVALUE,OUTKEY,OUTVALUE>
3.   void runNewMapper(final JobConf job,
4.                     final TaskSplitIndex splitIndex,
5.                     final TaskUmbilicalProtocol umbilical,
6.                     TaskReporter reporter
7.                     ) throws IOException,ClassNotFoundException,
8.                              InterruptedException{
9.     // make a task context so we can get the classes
10.     org.apache.hadoop.mapreduce.TaskAttemptContext taskContext =
11.       new org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl(job,
12.                                                                   getTaskID(),
13.                                                                   reporter);
14.     // make a mapper
15.     org.apache.hadoop.mapreduce.Mapper<INKEY,INVALUE,OUTKEY,OUTVALUE> mapper =
16.       (org.apache.hadoop.mapreduce.Mapper<INKEY,INVALUE,OUTKEY,OUTVALUE>)
17.         ReflectionUtils.newInstance(taskContext.getMapperClass(), job);
18.     // make the input format
19.     org.apache.hadoop.mapreduce.InputFormat<INKEY,INVALUE> inputFormat =
20.       (org.apache.hadoop.mapreduce.InputFormat<INKEY,INVALUE>)
21.         ReflectionUtils.newInstance(taskContext.getInputFormatClass(), job);
22.     // rebuild the input split
23.     org.apache.hadoop.mapreduce.InputSplit split = null;
24.     split = getSplitDetails(newPath(splitIndex.getSplitLocation()),
25.         splitIndex.getStartOffset());
26.     LOG.info("Processing split: "+ split);
27.  
28.     org.apache.hadoop.mapreduce.RecordReader<INKEY,INVALUE> input =
29.       newNewTrackingRecordReader<INKEY,INVALUE>
30.         (split, inputFormat, reporter, taskContext);
31.  
32.     job.setBoolean(JobContext.SKIP_RECORDS, isSkipping());
33.     org.apache.hadoop.mapreduce.RecordWriter output = null;
34.  
35.     // get an output object
36.     if(job.getNumReduceTasks()==0){
37.       output =  如果jreduce个数等于0.则执行该方法
38.         newNewDirectOutputCollector(taskContext, job, umbilical, reporter);
39.     }else{
40.        如果reduce个数大于0.则执行该方法
41.       output =newNewOutputCollector(taskContext, job, umbilical, reporter);
42.     }
43.  
44.     org.apache.hadoop.mapreduce.MapContext<INKEY, INVALUE, OUTKEY, OUTVALUE>
45.     mapContext =
46.       newMapContextImpl<INKEY, INVALUE, OUTKEY, OUTVALUE>(job, getTaskID(),
47.           input, output,
48.           committer,
49.           reporter, split);
50.  
51.     org.apache.hadoop.mapreduce.Mapper<INKEY,INVALUE,OUTKEY,OUTVALUE>.Context
52.         mapperContext =
53.           newWrappedMapper<INKEY, INVALUE, OUTKEY, OUTVALUE>().getMapContext(
54.               mapContext);
55.  
56.     try{
57.       input.initialize(split, mapperContext);
58.       mapper.run(mapperContext);
59.       mapPhase.complete();
60.       setPhase(TaskStatus.Phase.SORT);
61.       statusUpdate(umbilical);
62.       input.close();
63.       input = null;
64.       output.close(mapperContext);
65.       output = null;
66.     } finally {
67.       closeQuietly(input);
68.       closeQuietly(output, mapperContext);
69.     }
70.   }

我们知道，分区是在map函数输出的时候做的 ，所以这里是get output object

1. // get an output object
2.     if(job.getNumReduceTasks()==0){
3.  
4.       output =  如果jreduce个数等于0.则执行该方法
5.         newNewDirectOutputCollector(taskContext, job, umbilical, reporter);
6.     }else{
7.        如果reduce个数大于0.则执行该方法
8.       output =newNewOutputCollector(taskContext, job, umbilical, reporter);
9.     }

如果没有reduce任务，则new NewDirectOutputCollector()

（Collection过程我还没探索过呢）

如果有NewOutputCollector任务，则运行new NewOutputCollector()

 

内部类NewOutputCollector

在内部类NewOutputCollector中找到该方法（构造方法）

1. NewOutputCollector(org.apache.hadoop.mapreduce.JobContext jobContext,
2.                        JobConf job,
3.                        TaskUmbilicalProtocol umbilical,
4.                        TaskReporter reporter
5.                        ) throws IOException,ClassNotFoundException{
6.       collector = createSortingCollector(job, reporter);
7.  
8.       partitions = jobContext.getNumReduceTasks();
9.  
10.       if(partitions >1){
11.         partitioner =(org.apache.hadoop.mapreduce.Partitioner<K,V>)
12.           ReflectionUtils.newInstance(jobContext.getPartitionerClass(), job);
13.       }else{
14.         partitioner =new org.apache.hadoop.mapreduce.Partitioner<K,V>(){
15.           @Override
16.           publicint getPartition(K key, V value,int numPartitions){
17.             return partitions -1;
18.           }
19.         };
20.       }
21.     }

通过partitions = jobContext.getNumReduceTasks();语句获取到Reduce任务个数

如果Reduce任务数小于等于1，则新建一个Partitioner对象的同时并复写getPartition方法，这个复写的方法直接统一返回-1，就都在一个分区了。

如果Reduce任务数大于 ，则通过反射创建jobContext.getPartitionerClass()获取到的对象

于是查看：

jobContext接口

jobContext接口中的

1. /**
2.    * Get the {@link Partitioner} class for the job.
3.    *
4.    * @return the {@link Partitioner} class for the job.
5.    */
6.   publicClass<? extends Partitioner<?,?>> getPartitionerClass()
7.      throws ClassNotFoundException;

我们还是看其实现类jobContextImpl吧

jobContextImpl类

注意是在mapreduce包下啊，不是mapred包下

1. /**
2.    * Get the {@link Partitioner} class for the job.
3.    *
4.    * @return the {@link Partitioner} class for the job.
5.    */
6.   @SuppressWarnings("unchecked")
7.   publicClass<? extends Partitioner<?,?>> getPartitionerClass()
8.      throws ClassNotFoundException{
9.     return(Class<? extends Partitioner<?,?>>)
10.       conf.getClass(PARTITIONER_CLASS_ATTR,HashPartitioner.class);
11.   }

conf.getClass(PARTITIONER_CLASS_ATTR, HashPartitioner.class);

的意思是，从PARTITIONER_CLASS_ATTR属性中取出值，作为类返回，如果不存在，则使用和默认值HashPartitioner.class

也就是说，当Reduce个数大于1的时候，其默认调用的是HashPartitioner.class

1. publicclassHashPartitioner<K, V>extendsPartitioner<K, V>{
3. /** Use {@link Object#hashCode()} to partition. */
4. publicint getPartition(K key, V value,
5. int numReduceTasks){
6. return(key.hashCode()&Integer.MAX_VALUE)% numReduceTasks;
7. }
8. }

发现HashPartitioner调用的是getPartition方法，最终使用的是key对象中的hashcode方法

而我们使用eclipse（Alt+Shift+ S  按下H）复写的hashcode是将两个属性（账户和金额都考虑进去了）

嗯，果然自己修改自定义key类中的hashcode，测试了一下是可以的，只要hashcode是只根据我们的账户account进行生产

1. @Override
2.         publicint hashCode(){
3.             final int prime =31;
4.             int result =1;
5.             result = prime * result +((account == null)?0: account.hashCode());
6.      //     result = prime * result + ((amount == null) ? 0 : amount.hashCode());
7.             return result;
8.         }

 

另一种更主流的方式：

自定义的Partition类为什么要是Group的内部类呢？自己改为外部类自己测试下，发现完全可以

具体的形式

1. publicstaticclassKeyPartitioner extends  Partitioner<SelfKey,DoubleWritable>{
2.  
3.             @Override
4.             publicint getPartition(SelfKey key,DoubleWritable value,int numPartitions){
5.                 /**
6.                  * 如何保证数据整体输出上的有序，需要我们自定义业务逻辑
7.                  * 必须提示前知道num reduce task 个数？
8.                  * w  单词字符[a-zA-Z_0-9]
9.                  *  
10.                  */
11.                 String account =key.getAccount();
12.                 //0xxaaabbb 0-9 
13.                 //[0-2][3-6][7-9]
14.                 if(account.matches("\w*[0-2]")){
15.                     return0;
16.                 }elseif(account.matches("\w*[3-6]")){
17.                     return1;
18.                 }elseif(account.matches("\w*[7-9]")){
19.                     return2;
20.                 }
21.                 return0;
22.  
23.             }
24.         }

这是为了保证S1和S2都在分区1，而不会出现S1中的其中几个在分区1 ，另外几个在分区2

因为我们此时的键——是账户+金额，所以可能明明都是账户S1的分区却不一样，最后导致排序混乱？

 

 

 

来自为知笔记(Wiz)