5.4.10 分区

如果reduce任务数量大于1，也就是分区数量大于1，调用jobContext.getPartitionerClass()来获取分区类的Class对象。单击getPartitionerClass()->JobContext接口中：

xxxxxxxxxx
public Class<? extends Partitioner<?,?>> getPartitionerClass() 
     throws ClassNotFoundException;

Ctrl+Alt+B->JobContextImpl:

xxxxxxxxxx
 public Class<? extends Partitioner<?,?>> getPartitionerClass() 
     throws ClassNotFoundException {
    return (Class<? extends Partitioner<?,?>>) 
      conf.getClass(PARTITIONER_CLASS_ATTR, HashPartitioner.class);
  }

获取PARTITIONER_CLASS_ATTR常量的值mapreduce.job.partitioner.class,该字符串去job.xml找它对应的value值。如果没有定义，则使用默认的HashPartitioner类作为分区类。

xxxxxxxxxx
/** Partition keys by their {@link Object#hashCode()}. */
@InterfaceAudience.Public
@InterfaceStability.Stable
public class HashPartitioner<K, V> extends Partitioner<K, V> {

  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }
}

默认的分区的逻辑是，使用key的hashCode值%reduce任务数(也是分区总数) 求余数。

如何设置自定义的分区类：

xxxxxxxxxx
job.setPartitionerClass(WCPartitioner.class);
job.setNumReduceTasks(4);

setPartitionerClass()方法的代码：

xxxxxxxxxx
  public void setPartitionerClass(Class<? extends Partitioner> cls
                                  ) throws IllegalStateException {
    ensureState(JobState.DEFINE);
    conf.setClass(PARTITIONER_CLASS_ATTR, cls, 
                  Partitioner.class);
  }