注册 登录
> 编程技术

MapReduce的排序和二次排序

系统 1692 0
自己学习排序和二次排序的知识整理如下。
1.Hadoop的序列化格式介绍:Writable
2.Hadoop的key排序逻辑
3.全排序
4.如何自定义自己的Writable类型
5.如何实现二次排序

1.Hadoop的序列化格式介绍:Writable
要了解和编写MR实现排序必须要知道的第一个知识点就是Writable相关的接口和类，这些是HADOOP自己的序列化格式。更多的可能是要关注他的Subinterfaces：WritableComparable<T>。他是继承Writable和Comparable<T>接口，继而WritableComparable<T>的实现除了具有序列化特性，更重要的是具有了比较的特性，而比较的特性在MapReduce里是很重要的，因为MR中有个基于键的排序过程，所以可以作为键的类型必须具有Comparable<T>的特性。
除了WritableComparable接口外，还有一个接口RawComparaotor。
WritableComparable和RawComparator两个接口的区别是：
WritableComparable是需要把数据流反序列化为对象后，然后做对象之间的比较，而RawComparator是直接比较数据流的数据，不需要数据流反序列化成对象，省去了新建对象的开销。
2.Hadoop的key排序逻辑
Hadoop本身Key的数据类型的排序逻辑其实就是依赖于Hadoop本身的继承与WritableComparable<T>的基本数据类型和其他类型(相关类型可参考《Hadoop权威指南》第二版的90页)的compareTo方法的定义。
Key排序的规则：
1.如果调用jobconf的setOutputKeyComparatorClass()设置mapred.output.key.comparator.class
2.否则，使用key已经登记的comparator
3.否则，实现接口WritableComparable的compareTo()函数来操作
例如IntWritable的比较算法如下：
Java代码 收藏代码
  1. public int compareTo(Objecto){
  2. int thisValue= this .value;
  3. int thatValue=((IntWritable)o).value;
  4. return (thisValue<thatValue?- 1 :(thisValue==thatValue? 0 : 1 ));
  5. }
可以修改compareTo来实现自己所需的比较算法。

虽然我们知道是compareTo这个方法实现Key的排序，但其实我们在使用Hadoop的基本数据类型时不需要关注这个排序如何实现，因为Hadoop的框架会自动调用compareTo这个方法实现key的排序。但是这个排序只是局限在map或者reduce内部。针对于map与map，reduce与reduce之间的排序compareTo就管不着了，虽然这种情况不常出现，但是确实存在这种问题的，而且确实有适用场景，比如说全排序。
3.全排序
这里就需要关注Partition这个阶段，Partition阶段是针对每个Reduce，需要创建一个分区，然后把Map的输出结果映射到特定的分区中。这个分区中可能会有N个Key对应的数据，但是一个Key的所有数据只能在一个分区中。在实现全排序的过程中，如果只有一个reduce，也就是只有一个Partition，那么所有Map的输出都会经过一个Partition到一个reduce里，在一个reduce里可以根据compareTo(也可以采用其他比较算法)来排序，实现全排序。但是这种情况就让MapReduce失去了分布式计算的光环。
所以全排序的大概思路为：确保Partition之间是有序的就OK了，即保证Partition1的最大值小于Partition2的最小值就OK了，即便这样做也还是有个问题：Partition的分布不均，可能导致某些Partition处理的数据量远大于其他Partition处理的数据量。而实现全排序的核心步骤为：取样和Partition。
先“取样”，保证Partition得更均匀：
1) 对Math.min(10, splits.length)个split（输入分片）进行随机取样，对每个split取10000个样，总共10万个样
2) 10万个样排序，根据reducer的数量(n)，取出间隔平均的n-1个样
3) 将这个n-1个样写入partitionFile(_partition.lst，是一个SequenceFile)，key是取的样，值是nullValue
4) 将partitionFile写入DistributedCache
整个全排序的详细介绍可参照： http://www.iteye.com/topic/709986
4.如何自定义自己的Writable类型
自定义自己的Writable类型的场景应该很简单：Hadoop自带的数据类型要么在功能上不能满足需求，要么在性能上满足需求，毕竟Hadoop还在发展，不是所有情况都考虑的，但是他提供了自主的框架实现我们想要的功能。
定义自己的Writable类型需要实现：
a.重载构造函数
b.实现set和get方法
c.实现接口的方法：write()、readFields()、compareTo()
d.(可选)相当于JAVA构造的对象，重写java.lang.Object的hashCode()、equals()、toString()。Partition阶段默认的hashpartitioner会根据hashCode()来选择分区，如果不要对自定义类型做key进行分区，hashCode()可不实现
具体例子可参考hadoop的基本类型IntWritable的实现
Java代码 收藏代码
  1. public class IntWritable implements WritableComparable{
  2. private int value;
  4. public IntWritable(){}
  6. public IntWritable( int value){set(value);}
  8. /**SetthevalueofthisIntWritable.*/
  9. public void set( int value){ this .value=value;}
  11. /**ReturnthevalueofthisIntWritable.*/
  12. public int get(){ return value;}
  14. public void readFields(DataInputin) throws IOException{
  15. value=in.readInt();
  16. }
  18. public void write(DataOutputout) throws IOException{
  19. out.writeInt(value);
  20. }
  22. /**Returnstrueiff<code>o</code>isaIntWritablewiththesamevalue.*/
  23. public boolean equals(Objecto){
  24. if (!(o instanceof IntWritable))
  25. return false ;
  26. IntWritableother=(IntWritable)o;
  27. return this .value==other.value;
  28. }
  30. public int hashCode(){
  31. return value;
  32. }
  34. /**ComparestwoIntWritables.*/
  35. public int compareTo(Objecto){
  36. int thisValue= this .value;
  37. int thatValue=((IntWritable)o).value;
  38. return (thisValue<thatValue?- 1 :(thisValue==thatValue? 0 : 1 ));
  39. }
  41. public StringtoString(){
  42. return Integer.toString(value);
  43. }
  44. }
5.如何实现二次排序
二次排序的工作原理涉及到如下几方面：
a.创建key的数据类型，key要包括两次排序的元素
b.setPartitionerClass(Class<? extends Partitioner> theClass)
hadoop0.20.0以后的函数为setPartitionerClass
c.setOutputKeyComparatorClass(Class<? extends RawComparator> theClass)
hadoop0.20.0以后的函数为 setSortComparatorClass
d.setOutputValueGroupingComparator(Class<? extends RawComparator> theClass）
hadoop0.20.0以后的函数为 setGroupingComparatorClass

根据hadoop自己提供的example：org.apache.hadoop.examplesSecondarySort来说明二次排序具体是如何实现的.

SecondarySort实现IntPair、FirstPartitioner、FirstGroupingComparator、MapClass、Reduce这几个内部类，然后在main函数中调用。先说明下main函数中有哪些地方和普通的MR代码不同。
不同点是多了这两个set：
job.setPartitionerClass(FirstPartitioner.class);
设置自定义的Partition操作，在此是调用我们自定义的内部类 FirstPartitioner
job.setGroupingComparatorClass(FirstGroupingComparator.class);
设置哪些value进入哪些key的迭代器中，在此是调用自定义的内部类 FirstGroupingComparator

具体的操作逻辑为：

a.定义一个作为key的类型IntPair，在IntPair中有两个变量first、second，SecondarySort就是在对first排序后再对second再排序处理
b.定义分区函数类FirstPartitioner，Key的第一次排序。在FirstPartitioner实现如何处理key的first，把key对应的数据划分到不同的分区中。这样key中first相同的value会被放在同一个reduce中，在reduce中再做第二次排序
c(代码没有实现，其实内部是有处理).key比较函数类，key的第二次排序，是继承WritableComparator的一个比较器。 setSortComparatorClass可以实现。
为什么没有使用 setSortComparatorClass()是因为hadoop对key排序的规则(参看 2.Hadoop的key排序逻辑 )决定的。由于我们在IntPair中已经定义了compareTo()函数。
d.定义分组函数类 FirstGroupingComparator， 保证只要key的的第一部分相同，value就进入key的value迭代器中。

MapReduce的排序和二次排序


更多文章、技术交流、商务合作、联系博主

微信扫码或搜索：z360901061

微信扫一扫加我为好友

QQ号联系： 360901061

您的支持是博主写作最大的动力，如果您喜欢我的文章，感觉我的文章对您有帮助，请用微信扫描下面二维码支持博主2元、5元、10元、20元等您想捐的金额吧，狠狠点击下面给点支持吧，站长非常感激您！手机微信长按不能支付解决办法：请将微信支付二维码保存到相册，切换到微信，然后点击微信右上角扫一扫功能，选择支付二维码完成支付。

【本文对您有帮助就好】元

您的支持是博主写作最大的动力，如果您喜欢我的文章，感觉我的文章对您有帮助，请用微信扫描上面二维码支持博主2元、5元、10元、自定义金额等您想捐的金额吧，站长会非常 感谢您的哦！！！

发表我的评论
最新评论 总共0条评论