Hadoop（5）Mapreduce 排序：次排序（Secondary sort）

45tr

4.2 排序（SORT）

在MapReduce中，排序的目的有两个：

• MapReduce可以通过排序将Map输出的键分组。然后可以对每一个的唯一键调用一次reduce。
• 在某些需要排序的特定场景中，用户可以将作业（job）的输出进行排序。
  

举例：需要了解前N个最受欢迎的用户或网页的数据分析工作。

在这一节中，有两个场景需要对MapReduce的排序行为进行优化。

• 次排序（Secondary sort）
• 全体顺序排序（Total order sort）
  

次排序允许基于reduce的键对它的值进行排序。如果要求一些数据先于另外一些数据到达reduce，次排序就很有用。（这一章在讲解优化过的重分区连接也提到了这样的场景。）另一个场景中，需要将作业的输出根据两个键进行排序，一个键的优先级高于另外一个键（secondary key）。这里也可以用到次排序。举例：将股票数据先根据股票标志进行主排序（primary sort），然后根据股票配额进行次排序。本书很多技术中将会运用次排序，如重分区连接的优化，朋友图算法等。

这一节第二部分中，将探讨对reduce的输出的全体数据进行排序。这在分析数据集中的前N个元素或后N个元素时会比较有用。

4.2.1 次排序（Secondary sort）

在前一节（MapReduce连接）中，次排序用于使一部分数据先于另外一部分到达reduce。作为基础知识，了解MapReduce中的数据整理和数据流是学习次排序前提。图4.12说明了三个影响数据整理和数据流（分区，排序，分组）的元素，并且说明了这些元素如何整合到MapReduce中。

在map输出收集（output collection）阶段，由分区器（Partitioner）选择哪个reduce应该接收map的输出。map输出的各个分区的数据，由RawComparator进行排序。Reduce端也用RawComparator进行排序。然后，由RawComparator对排序好的数据进行分组。

技术21 实现一个次排序

对于某个map的键的所有值，如果需要其中一部分值先于另外一部分值到达reduce，就可以用到次排序。在本书的第7章中的朋友图算法，和经过优化的重分区排序中，次排序就显得很有价值。

问题

在发送给某个reduce的数据中，需要对某个自然键（natural key）的值进行排序。

方案

这个技术中将应用到自定义分区类，排序比较类（sort comparator），分组比较类（grouping comparator）。这些是实现次排序的基础。

讨论

在这个技术中，使用次排序来对人的名字进行排序。具体步骤是：先用主排序对人的姓排序，再用次排序对人的名字排序。

次排序需要在map函数生成组合输出键（composite key）。

组合输出键包括两个部分：

• 自然键，用于连接。
• 二级键（secondary key），用于对隶属于自然键的值进行排序。排序后的结果将被发送给reduce。
  

图4.13说明了组合键的构成。它还包括了一个用于reduce端的组合值（composite value）。组合值让reduce可以访问二级键。

在完成好了组合键类之后，接下来具体说明分区，排序和分组阶段以及他们的实现。

组合键（COMPOSITE KEY）

组合键包括姓和名字。它扩展了WritableComparable。WritableComparable被推荐用于map函数输出键的Writable类。

public class Person implements WritableComparable<Person> {

    private String firstName;
    private String lastName;

    @Override
    public void readFields(DataInput in) throws IOException {
        this.firstName = in.readUTF();
        this.lastName = in.readUTF();
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(firstName);
        out.writeUTF(lastName);
    }
...

图4.14说明了分区，排序和分组的类的名字和方法的设置。同时还有各个类如何使用组合键。

接下来是对其他类的实现代码的介绍。

分区器（PARTITIONER）

分区器用来决定map的输出值应该分配到哪个reduce。MapReduce的默认分区器（HashPartitioner）调用输出键的hashCode方法，然后用hashCode方法的结果对reduce的数量进行一个模数（modulo）运算，最后得到那个目标reduce。默认的分区器使用整个键。这就不适用与组合键了。因为它可能把有同样自然键的组合键发送给不同的reduce。因此，就需要自定义分区器，基于自然键进行分区。

以下代码实现了分区器的接口。getPartition方法的输入参数有key，value和分区的数量：



1 public interface Partitioner<K2, V2> extends JobConfigurable {
2     int getPartition(K2 key, V2 value, int numPartitions);
3 }


自定义的分区器将基于Person类中的姓计算哈希值，然后将这个哈希值对分区的数量进行模运算。在这里，分区的数量就是reduce的数量：
public class PersonNamePartitioner extends Partitioner<Person, Text> {

    @Override
    public int getPartition(Person key, Text value, int numPartitions) {
        return Math.abs(key.getLastName().hashCode() * 127) % numPartitions;
    }

}

排序（SORTING）

Map端和reduce端都要进行排序。Map端排序的目的是让reduce端的排序更加高效。这里将让MapReduce使用整个组合键进行排序，也就是基于姓和名字。

在下列例子中实现了WritableComparator。WritableComparator基于姓和名对用户进行比较。

public class PersonComparator extends WritableComparator {

    protected PersonComparator() {
        super(Person.class, true);
    }

    @Override
    public int compare(WritableComparable w1, WritableComparable w2) {

        Person p1 = (Person) w1;
        Person p2 = (Person) w2;

        int cmp = p1.getLastName().compareTo(p2.getLastName());

        if (cmp != 0) {
            return cmp;
        }

        return p1.getFirstName().compareTo(p2.getFirstName());
    }
}

分组（GROUPING）

当reduce阶段将在本地磁盘上的map输出的记录进行流化处理（streaming）的时候，就要进行分组。在分组中，记录将被按一定方式排多个有逻辑顺序的记录流，并被传输给reduce。

在分组阶段，所有的记录已经经过了次排序。分组比较器需要将有相同姓的记录分在同一个组。下面是分组比较器的实现：

public class PersonNameComparator extends WritableComparator {

    protected PersonNameComparator() {
        super(Person.class, true);
    }

    @Override
    public int compare(WritableComparable o1, WritableComparable o2) {
        Person p1 = (Person) o1;
        Person p2 = (Person) o2;
        return p1.getLastName().compareTo(p2.getLastName());
    }
}

MAPREDUCE

最后一步是告诉MapReduce使用自定义的分区器类，排序比较器类和分组比较器类：



1 job.setPartitionerClass(PersonNamePartitioner.class);
2 job.setSortComparatorClass(PersonComparator.class);
3 job.setGroupingComparatorClass(PersonNameComparator.class);


然后需要实现map和reduce代码。Map类创建具有姓和名的组合键，然后将它作为输出键。将名字作为输出值。

Reduce类的输出和输入一样：

public static class Map extends Mapper<Text, Text, Person, Text> {

    private Person outputKey = new Person();

    @Override
    protected void map(Text lastName, Text firstName, Context context)
        throws IOException, InterruptedException {

        outputKey.set(lastName.toString(), firstName.toString());
        context.write(outputKey, firstName);

    }
}

public static class Reduce extends Reducer<Person, Text, Text, Text> {

    Text lastName = new Text();

    @Override
    public void reduce(Person key, Iterable<Text> values, Context context)
        throws IOException, InterruptedException {

        lastName.set(key.getLastName());

        for (Text firstName : values) {
            context.write(lastName, firstName);
        }
    }
}

上传一个名字是乱序的小文件，并测试次排序是否能够生成已经根据名字排序好的结果：

$ hadoop fs -put test-data/ch4/usernames.txt .

$ hadoop fs -cat usernames.txt
Smith John
Smith Anne
Smith Ken

$ bin/run.sh com.manning.hip.ch4.sort.secondary.SortMapReduce usernames.txt output

$ hadoop fs -cat output/part*
Smith Anne
Smith John
Smith Ken

上面的结果和期望一致。



小结

这一节展示了MapReduce中如何使用次排序。接下来是如何将多个reduce的结果整合起来排序。