[转]Hadoop MapReduce 二次排序原理及其应用

转载自:《Hadoop MapReduce 二次排序原理及应用》

关于二次排序主要涉及到这么几个东西:

0.20.0以前使用的是

  • setPartitionerClass
  • setOutputkeyComparatorClass
  • setOutputValueGroupingComparator

0.20.0以后使用是

  • job.setPartitionerClass(Partitioner p);
  • job.setSortComparatorClass(RawComparator c);
  • job.setGroupingComparatorClass(RawComparator c);

下面的例子里面只用到了 setGroupingComparatorClass

http://blog.csdn.net/heyutao007/article/details/5890103
mr自带的例子中的源码SecondarySort,我重新写了一下,基本没变。
这个例子中定义的map和reduce如下,关键是它对输入输出类型的定义:(java泛型编程)

1、首先说一下工作原理

在map阶段,使用job.setInputFormatClass定义的InputFormat将输入的数据集分割成小数据块splites,同时InputFormat提供一个RecordReder的实现。本例子中使用的是TextInputFormat,他提供的RecordReder会将文本的字节偏移量作为key,这一行的文本作为value。这就是自定义Map的输入是<LongWritable, Text>的原因。然后调用自定义Map的map方法,将一个个<LongWritable, Text>对输入给Map的map方法。注意输出应该符合自定义Map中定义的输出<IntPair, IntWritable>。最终是生成一个List<IntPair, IntWritable>。在map阶段的最后,会先调用job.setPartitionerClass对这个List进行分区,每个分区映射到一个reducer。每个分区内又调用job.setSortComparatorClass设置的key比较函数类排序。可以看到,这本身就是一个二次排序。 如果没有通过job.setSortComparatorClass设置key比较函数类,则使用key的实现的compareTo方法。 在第一个例子中,使用了IntPair实现的compareTo方法,而在下一个例子中,专门定义了key比较函数类。

在reduce阶段,reducer接收到所有映射到这个reducer的map输出后,也是会调用job.setSortComparatorClass设置的key比较函数类对所有数据对排序。然后开始构造一个key对应的value迭代器。这时就要用到分组,使用jobjob.setGroupingComparatorClass设置的分组函数类。只要这个比较器比较的两个key相同,他们就属于同一个组,它们的value放在一个value迭代器,而这个迭代器的key使用属于同一个组的所有key的第一个key。最后就是进入Reducer的reduce方法,reduce方法的输入是所有的(key和它的value迭代器)。同样注意输入与输出的类型必须与自定义的Reducer中声明的一致。

2、二次排序

就是首先按照第一字段排序,然后再对第一字段相同的行按照第二字段排序,注意不能破坏第一次排序 的结果 。例如 :

echo “3 b
1 c
2 a
1 d
3 a”|sort -k1 -k2
1 c
1 d
2 a
3 a
3 b

3、具体步骤

3.1 自定义key

在mr中,所有的key是需要被比较和排序的,并且是二次,先根据partitione,再根据大小。而本例中也是要比较两次。先按照第一字段排序,然后再对第一字段相同的按照第二字段排序。根据这一点,我们可以构造一个复合类IntPair,他有两个字段,先利用分区对第一字段排序,再利用分区内的比较对第二字段排序。
所有自定义的key应该实现接口WritableComparable,因为是可序列的并且可比较的。并重载方法

另外新定义的类应该重写的两个方法

由于key是自定义的,所以还需要自定义一下类。

3.2 分区函数类。这是key的第一次比较

在job中设置使用setPartitionerClasss

3.2 key比较函数类

这是key的第二次比较。这是一个比较器,需要继承WritableComparator。

必须有一个构造函数,并且重载 public int compare(WritableComparable w1, WritableComparable w2)

另一种方法是 实现接口RawComparator。
在job中设置使用setSortComparatorClass

3.3 分组函数类

在reduce阶段,构造一个key对应的value迭代器的时候,只要first相同就属于同一个组,放在一个value迭代器。这是一个比较器,需要继承WritableComparator。

同key比较函数类,必须有一个构造函数,并且重载 public int compare(WritableComparable w1, WritableComparable w2)

同key比较函数类,分组函数类另一种方法是实现接口RawComparator。
在job中设置使用setGroupingComparatorClass。

另外注意的是,如果reduce的输入与输出不是同一种类型,则不要定义Combiner也使用reduce,因为Combiner的输出是reduce的输入。除非重新定义一个Combiner。

4 代码

这个例子中没有使用key比较函数类,而是使用key的实现的compareTo方法:

7 原理图(点击查看大图)

003757_7qZb_568818

8、推荐阅读

hive中使用标准sql实现分组内排序

http://superlxw1234.iteye.com/blog/1869612

Pig、Hive、MapReduce 解决分组 Top K 问题

http://my.oschina.net/leejun2005/blog/85187

9、REF

mapreduce的二次排序 SecondarySort

http://blog.csdn.net/zyj8170/article/details/7530728

学会定制MapReduce里的partition,sort和grouping,Secondary Sort Made Easy 进行二次排序

http://blog.sina.com.cn/s/blog_9bf980ad0100zk7r.html

Simple Moving Average, Secondary Sort, and MapReduce (Part 3)

http://blog.cloudera.com/blog/2011/04/simple-moving-average-secondary-sort-and-mapreduce-part-3/

https://github.com/jpatanooga/Caduceus/tree/master/src/tv/floe/caduceus/hadoop/movingaverage

MapReduce的排序和二次排序原理总结

http://hugh-wangp.iteye.com/blog/1491175

泛型value的二次排序

http://wenku.baidu.com/view/a3826a235901020207409c47.html

http://vangjee.wordpress.com/2012/03/20/secondary-sorting-aka-sorting-values-in-hadoops-mapreduce-programming-paradigm/

Leave a Reply

Your email address will not be published.