Partition分区

　　分区是MapReduce框架的Map阶段进行数据处理之后，将数据写出时需要进行的一项操作，分区的数量决定了ReduceTask的数量，也决定了最终的输出文件有多少个。其中，Hadoop是有默认的分区方法的，即HashPartitioner类是默认的分区类，该类的源码如下：

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;
  }

}

　　该方法进行分区的时候，关键因素在于map输出数据的key的hashcode值，以及numReduceTasks的大小，其中numReduceTasks默认值是1，那么任何数模除1都是0，也就是说系统默认的分区数量只有一个，map输出的数据都会被汇总到一个分区里边，然后交由一个ReduceTask进行处理，最终的输出文件也就只有一个。而对于不同类型的key，对应的hashcode值计算方法也不一样，比如，key是Integer类型时，那么它的hashcode值就是key这个值本身，源码如下：

    public int hashCode() {
        return Integer.hashCode(value);
    }

    public static int hashCode(int value) {
        return value;
    }

　　如果是字符串类型，那么hashcode值就是s[0]*31^(n-1) + s[1]*31^(n-2) + ... + s[n-1]，这里假设字符串s，长度为n，源码如下：

    public int hashCode() {
        int h = hash;
        if (h == 0 && value.length > 0) {
            char val[] = value;

            for (int i = 0; i < value.length; i++) {
                h = 31 * h + val[i];
            }
            hash = h;
        }
        return h;
    }

　　至于key.hashcode()和Integer.MAX_VALUE进行“与”运算，是为了保证进行模运算的值在Integer类型（32位整型数）的合理范围内，也避免出现负数（因为分区号的值只能从0开始，不能出现负数）。Integer.MAX_VALUE的值为0x7fffffff，它会将key.hashcode这个值中多余的部分去掉，从而是一个合理的数值。

　　那么当我们想根据自己的实际业务将统计结果输出到不同的文件中呢？系统默认的分区模式肯定不行，因为默认的方式将统计结果全部输出到一个文件中，当然，如果是单纯的设置分区数量，那么也是不行的，因为不同文件可能对应不同的条件。这就需要通过自定义分区来解决这些问题了。具体操作如下：

　　自定义一个分区类，然后继承Partitioner这个类,这个类是一个抽象类，自定义的分区类都要继承它：

public abstract class Partitioner<KEY, VALUE> {
  
  /** 
   * Get the partition number for a given key (hence record) given the total 
   * number of partitions i.e. number of reduce-tasks for the job.
   *   
   * <p>Typically a hash function on a all or a subset of the key.</p>
   *
   * @param key the key to be partioned.
   * @param value the entry value.
   * @param numPartitions the total number of partitions.
   * @return the partition number for the <code>key</code>.
   */
  public abstract int getPartition(KEY key, VALUE value, int numPartitions);
  
}

　　继承这个类的时候要指定一下这个抽象类的泛型类型，这个类型是和map方法写出的key和value对应一致，继承这个类最主要的是实现getPartition这个方法，通过注释我们知道这个方法的任务是：获取给定key的分区号，给定总的分区数，即reducemask的数量，所以它的返回值就是当前key的分区号。

　　这里我们的例子是，根据手机号前三位将手机号划分为不同的地区进行统计，并输出到不同的文件（分区），自定义的分区类如下：

public class ProvincePartitioner extends Partitioner<Text,FlowBean> {

    @Override
    public int getPartition(Text key, FlowBean flowBean, int numPartitions) {
        // key是手机号
        // value是流量信息

        // 获取手机号前三位
        String prePhoneNum = key.toString().substring(0, 3);
        int partition = 4;
        if ("136".equals(prePhoneNum)){
            partition = 0;
        }else if ("137".equals(prePhoneNum)){
            partition = 1;
        }else if ("138".equals(prePhoneNum)){
            partition = 2;
        }else if ("139".equals(prePhoneNum)){
            partition = 3;
        }
        return partition;
    }
}

　　自定义完成之后，需要在驱动类（一个MapReduce程序要包括三个部分Mapper、Reducer、xxxDriver）里面进行设置，

job.setPartitionerClass(ProvincePartitioner.class);
// 设置好使用自定义的Partitioner，必须设置NumReduceTasks，否则还会按照默认的1来只开启一个
// reducetask进程，输出的分区文件也只有一个
// 如果自定义分区的个数小于设置的NumReduceTasks值，会抛出异常
job.setNumReduceTasks(5);

　　第一行代码是指定使用自定义的分区类，第二行代码是指定分区总数量，其中特别需要注意这里指定分区的总数量，既是最终输出文件中的数量，也是我们在getPartition(KEY key, VALUE value, int numPartitions);这个方法中，第二个参数的值，所以不能随便指定：

　　1、如果NumReduceTasks > getPartition返回值，则会产生多余的输出文件，多几个看他俩差值是几个；

　　2、如果 1 < NumReduceTasks < getPartition返回值，则会发生异常，因为有一部分分区数据无处安放；

　　3、如果 1 == NumReduceTasks，则不管Maptask输出多少个分区文件，最终结果都会汇集到一个ReduceTask，最终的输出文件也是一个

还有就是，分区号必须从0开始，依次累加递增。