Partitioner分区过程分析

            Partition中国人意味着分区，意义的碎片，这个阶段也是整个MapReduce该过程的第三阶段。在Map返回任务，是使key分到通过一定的分区算法。分到固定的区域中。给不同的Reduce做处理，达到负载均衡的目的。

他的运行过程事实上就是发生在上篇文章提到的collect的过程阶段，当输入的key调用了用户的map函数时。中间结果就会被分区了。虽说这个过程看似不是非常重要，可是也有值得学习的地方。Hadoop默认的算法是HashPartitioner，就是依据key的hashcode取摸运算，非常easy的。

/** Partition keys by their {@link Object#hashCode()}. 
 */
public class HashPartitioner<K2, V2> implements Partitioner<K2, V2> {

  public void configure(JobConf job) {}

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

}

先看看关系Partition的相关类结构。

可见。TotalOrderPartitioner还是挺复杂的。

        TotalOrderPartitioner的作用就是保证全局有序，对于key的划分，他划分了几个key的抽样点。作为key的划分点，比【2,4,6,8】，4个key抽样点。把区间划成了5份，假设某个key的值为5,他的区间为4-6。所以在第三区间，也就是说，这个类的作用就是环绕给定的划分点，寻找他的区间号，就代表任务的完毕，至于你中间用的是二分搜索。还是其它的什么算法。都由你说了算。

      好的。首先第一步，从配置文件里得到划分点，他事实上是存在于一个叫partition.file的文件里，配置中仅仅保留了路径，

public void configure(JobConf job) {
    try {
      //获得partition file
      String parts = getPartitionFile(job);
      final Path partFile = new Path(parts);
      final FileSystem fs = (DEFAULT_PATH.equals(parts))
        ? FileSystem.getLocal(job)     // assume in DistributedCache
        : partFile.getFileSystem(job);

      Class<K> keyClass = (Class<K>)job.getMapOutputKeyClass();
      //从partition中读出Spilts分区点
      K[] splitPoints = readPartitions(fs, partFile, keyClass, job);
      ....

       然后開始关键的操作了，假设你的key值类型不是BinaryComparable二进制比較类型的话。比方能直接比較值的数字类型，就直接用二分算法。创建二分搜索节点。传入自己的比較器实现:

....
      RawComparator<K> comparator =
        (RawComparator<K>) job.getOutputKeyComparator();
      for (int i = 0; i < splitPoints.length - 1; ++i) {
        if (comparator.compare(splitPoints[i], splitPoints[i+1]) >= 0) {
          throw new IOException("Split points are out of order");
        }
      }
      boolean natOrder =
        job.getBoolean("total.order.partitioner.natural.order", true);
      //推断是否为BinaryComparable类型，假设是，建立Trie树
      if (natOrder && BinaryComparable.class.isAssignableFrom(keyClass)) {
        partitions = buildTrie((BinaryComparable[])splitPoints, 0,
            splitPoints.length, new byte[0],
            job.getInt("total.order.partitioner.max.trie.depth", 2));
      } else {
    	//假设是不是则建立构建BinarySearchNode，用二分查找。用自己构建的比較器
        partitions = new BinarySearchNode(splitPoints, comparator);
      }

/**
   * For types that are not {@link org.apache.hadoop.io.BinaryComparable} or
   * where disabled by <tt>total.order.partitioner.natural.order</tt>,
   * search the partition keyset with a binary search.
   */
  class BinarySearchNode implements Node<K> {
	//比較的内容节点
    private final K[] splitPoints;
    //比較器
    private final RawComparator<K> comparator;
    BinarySearchNode(K[] splitPoints, RawComparator<K> comparator) {
      this.splitPoints = splitPoints;
      this.comparator = comparator;
    }
    
   /**
    * 通过自己传入的比較器方法进行二分查找
    */
    public int findPartition(K key) {
      final int pos = Arrays.binarySearch(splitPoints, key, comparator) + 1;
      return (pos < 0) ?
 -pos : pos;
    }
  }

/**
   * An inner trie node that contains 256 children based on the next
   * character.
   */
  class InnerTrieNode extends TrieNode {
    private TrieNode[] child = new TrieNode[256];

    InnerTrieNode(int level) {
      super(level);
    }
    ...

private TrieNode buildTrie(BinaryComparable[] splits, int lower,
      int upper, byte[] prefix, int maxDepth) {
    final int depth = prefix.length;
    if (depth >= maxDepth || lower == upper) {
      //深度抵达最大的时候，应创建叶子节点了
      return new LeafTrieNode(depth, splits, lower, upper);
    }
    InnerTrieNode result = new InnerTrieNode(depth);
    byte[] trial = Arrays.copyOf(prefix, prefix.length + 1);
    // append an extra byte on to the prefix
    int currentBound = lower;
    //每一个父节点拥有着255个子节点
    for(int ch = 0; ch < 255; ++ch) {
      trial[depth] = (byte) (ch + 1);
      lower = currentBound;
      while (currentBound < upper) {
        if (splits[currentBound].compareTo(trial, 0, trial.length) >= 0) {
          break;
        }
        currentBound += 1;
      }
      trial[depth] = (byte) ch;
      //result.child为首节点。递归创建子节点
      result.child[0xFF & ch] = buildTrie(splits, lower, currentBound, trial,
                                   maxDepth);
    }
    // pick up the rest
    trial[depth] = 127;
    result.child[255] = buildTrie(splits, currentBound, upper, trial,
                                  maxDepth);
    return result;
  }

....
    /**
     * 非叶子的节点的查询
     */
    public int findPartition(BinaryComparable key) {
      //获取当前的深度
      int level = getLevel();
      
      if (key.getLength() <= level) {
        return child[0].findPartition(key);
      }
      
      //从key在此位置相应的字符child開始继续搜寻下一个，key.getBytes()[level]为第level位置的字符
      return child[0xFF & key.getBytes()[level]].findPartition(key);
    }

....
    //在叶子节点，进行二分查找分区号
    public int findPartition(BinaryComparable key) {
      final int pos = Arrays.binarySearch(splitPoints, lower, upper, key) + 1;
      return (pos < 0) ? -pos : pos;
    }