hadoop二次排序的原理和实现方法(2)

在本例中要比较两次。先按照第一字段排序，然后再对第一字段相同的按照第二字段排序。根据这一点，我们可以构造一个复合类IntPair ，它有两个字段，先利用分区对第一字段排序，再利用分区内的比较对第二字段排序。二次排序的流程分为以下几步。

1、自定义 key

所有自定义的key应该实现接口WritableComparable，因为它是可序列化的并且可比较的。WritableComparable 的内部方法如下所示

// 反序列化，从流中的二进制转换成IntPair
public void readFields(DataInput in) throws IOException

// 序列化，将IntPair转化成使用流传送的二进制
public void write(DataOutput out)

// key的比较
public int compareTo(IntPair o)

// 默认的分区类 HashPartitioner，使用此方法
public int hashCode()

// 默认实现
public boolean equals(Object right)

2、自定义分区

自定义分区函数类FirstPartitioner，是key的第一次比较，完成对所有key的排序。

public static class FirstPartitioner extends Partitioner< IntPair,IntWritable>

在job中使用setPartitionerClasss()方法设置Partitioner

job.setPartitionerClasss(FirstPartitioner.Class);

3、Key的比较类

这是Key的第二次比较，对所有的Key进行排序，即同时完成IntPair中的first和second排序。该类是一个比较器，可以通过两种方式实现。

1) 继承WritableComparator。

public static class KeyComparator extends WritableComparator

必须有一个构造函数，并且重载以下方法。

public int compare(WritableComparable w1, WritableComparable w2)

2) 实现接口 RawComparator。

上面两种实现方式，在Job中，可以通过setSortComparatorClass()方法来设置Key的比较类。

job.setSortComparatorClass(KeyComparator.Class);

注意：如果没有使用自定义的SortComparator类，则默认使用Key中compareTo()方法对Key排序。

4、定义分组类函数

在Reduce阶段，构造一个与 Key 相对应的 Value 迭代器的时候，只要first相同就属于同一个组，放在一个Value迭代器。定义这个比较器，可以有两种方式。

1) 继承 WritableComparator。

public static class GroupingComparator extends WritableComparator

必须有一个构造函数，并且重载以下方法。

public int compare(WritableComparable w1, WritableComparable w2)

2) 实现接口 RawComparator。

上面两种实现方式，在 Job 中，可以通过 setGroupingComparatorClass()方法来设置分组类。

job.setGroupingComparatorClass(GroupingComparator.Class);

另外注意的是，如果reduce的输入与输出不是同一种类型，则 Combiner和Reducer 不能共用 Reducer 类，因为