[大牛翻译系列]Hadoop（5）MapReduce 排序：次排序（Secondary sort）

371918642

4.2 排序（SORT）
　　在MapReduce中，排序的目的有两个：

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

　　例如：需要了解前N个最受欢迎的用户或网页的数据分析工作。
　　在这一节中，有两个场景需要对MapReduce的排序行为进行优化。

• 次排序（Secondary sort）
  
• 总排序（Total order sorting）
  

　　次排序可以根据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类。
　　



1 public class Person implements WritableComparable {
2
3     private String firstName;
4     private String lastName;
5     
6     @Override
7     public void readFields(DataInput in) throws IOException {
8         this.firstName = in.readUTF();
9         this.lastName = in.readUTF();
10     }
11     
12     @Override
13     public void write(DataOutput out) throws IOException {
14         out.writeUTF(firstName);
15         out.writeUTF(lastName);
16     }
17 ...
　　
　　图4.14说明了分区，排序和分组的类的名字和方法的设置。同时还有各个类如何使用组合键。
　　

　　
　　接下来是对其它类的实现代码的介绍。
　　
　　分区器（PARTITIONER）
　　分区器用来决定map的输出值应该分配到哪个reduce。MapReduce的默认分区器（HashPartitioner）调用输出键的hashCode方法，然后用hashCode方法的结果对reduce的数量进行一个模数（modulo）运算，最后得到那个目标reduce。默认的分区器使用整个键。这就不适于组合键了。因为它可能把有同样自然键的组合键发送给不同的reduce。因此，就需要自定义分区器，基于自然键进行分区。
　　以下代码实现了分区器的接口。getPartition方法的输入参数有key，value和分区的数量：
　　



1 public interface Partitioner extends JobConfigurable {
2     int getPartition(K2 key, V2 value, int numPartitions);
3 }
　　
　　自定义的分区器将基于Person类中的姓计算哈希值，然后将这个哈希值对分区的数量进行模运算。在这里，分区的数量就是reduce的数量：
　　



1 public class PersonNamePartitioner extends Partitioner {
2
3     @Override
4     public int getPartition(Person key, Text value, int numPartitions) {
5         return Math.abs(key.getLastName().hashCode() * 127) % numPartitions;
6     }
7
8 }
　　
　　排序（SORTING）
　　Map端和reduce端都要进行排序。Map端排序的目的是让reduce端的排序更加高效。这里将让MapReduce使用组合键的所有值进行排序，也就是基于姓氏和名字。
　　在下列例子中实现了WritableComparator。WritableComparator比较用户的姓氏和名字。
　　



1 public class PersonComparator extends WritableComparator {
2
3     protected PersonComparator() {
4         super(Person.class, true);
5     }
6
7     @Override
8     public int compare(WritableComparable w1, WritableComparable w2) {
9     
10         Person p1 = (Person) w1;
11         Person p2 = (Person) w2;
12         
13         int cmp = p1.getLastName().compareTo(p2.getLastName());
14         
15         if (cmp != 0) {
16             return cmp;
17         }
18         
19         return p1.getFirstName().compareTo(p2.getFirstName());
20     }
21 }
　　
　　分组（GROUPING）
　　当reduce阶段将在本地磁盘上的map输出的记录进行流化处理（streaming）的时候，需要要进行分组。在分组中，记录将被按一定方式排成一个有逻辑顺序的流，并被传输给reduce。
　　在分组阶段，所有的记录已经经过了次排序。分组比较器需要将有相同姓氏的记录分在同一个组。下面是分组比较器的实现：
　　



1 public class PersonNameComparator extends WritableComparator {
2
3     protected PersonNameComparator() {
4         super(Person.class, true);
5     }
6     
7     @Override
8     public int compare(WritableComparable o1, WritableComparable o2) {
9         Person p1 = (Person) o1;
10         Person p2 = (Person) o2;
11         return p1.getLastName().compareTo(p2.getLastName());
12     }
13 }
　　
　　MAPREDUCE
　　最后一步是告诉MapReduce使用自定义的分区器类，排序比较器类和分组比较器类：
　　



1 job.setPartitionerClass(PersonNamePartitioner.class);
2 job.setSortComparatorClass(PersonComparator.class);
3 job.setGroupingComparatorClass(PersonNameComparator.class);
　　
　　然后需要实现map和reduce代码。Map类创建具有姓和名的组合键，然后将它作为输出键。将名字作为输出值。
　　Reduce类的输出和输入一样：
　　



1 public static class Map extends Mapper {
2
3     private Person outputKey = new Person();
4     
5     @Override
6     protected void map(Text lastName, Text firstName, Context context)
7         throws IOException, InterruptedException {
8         
9         outputKey.set(lastName.toString(), firstName.toString());
10         context.write(outputKey, firstName);
11         
12     }
13 }
14
15 public static class Reduce extends Reducer {
16
17     Text lastName = new Text();
18     
19     @Override
20     public void reduce(Person key, Iterable values, Context context)
21         throws IOException, InterruptedException {
22         
23         lastName.set(key.getLastName());
24         
25         for (Text firstName : values) {
26             context.write(lastName, firstName);
27         }
28     }
29 }
　　
　　上传一个包含了乱序的名字的小文件，并测试次排序是否能够生成已经根据名字排序好的结果：
　　



$ 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的结果做总体排序。