【原】MapReduce的排序和二次排序

标签: mapreduce 排序 排序 | 发表时间:2012-04-20 10:38 | 作者:
出处:http://www.iteye.com

自己学习排序和二次排序的知识整理如下。
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的比较算法如下:
public int compareTo(Object o) {
    int thisValue = this.value;
    int thatValue = ((IntWritable)o).value;
    return (thisValue<thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
  }
 
可以修改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的实现
public class IntWritable implements WritableComparable {
  private int value;

  public IntWritable() {}

  public IntWritable(int value) { set(value); }

  /** Set the value of this IntWritable. */
  public void set(int value) { this.value = value; }

  /** Return the value of this IntWritable. */
  public int get() { return value; }

  public void readFields(DataInput in) throws IOException {
    value = in.readInt();
  }

  public void write(DataOutput out) throws IOException {
    out.writeInt(value);
  }

  /** Returns true iff <code>o</code> is a IntWritable with the same value. */
  public boolean equals(Object o) {
    if (!(o instanceof IntWritable))
      return false;
    IntWritable other = (IntWritable)o;
    return this.value == other.value;
  }

  public int hashCode() {
    return value;
  }

  /** Compares two IntWritables. */
  public int compareTo(Object o) {
    int thisValue = this.value;
    int thatValue = ((IntWritable)o).value;
    return (thisValue<thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
  }

  public String toString() {
    return Integer.toString(value);
  }
}
 

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迭代器中。


已有 0 人发表留言,猛击->> 这里<<-参与讨论


ITeye推荐



相关 [mapreduce 排序 排序] 推荐:

MapReduce之二次排序

- - CSDN博客云计算推荐文章
     面试官问了一个MapReduce问题:“如何用MapReduce实现两个表的连接”.      我说“用两个job实现,第一个. 于是这两天回来看了下MapReduce的二次排序.      在某些情况下,需要对reduce中的value进行排序. 二次排序,可以将根据key聚合起来的valueList根据value进行排序.

mapreduce倒排序索引

- - CSDN博客云计算推荐文章
  private Text word_filepath = new Text();//文件路径.   private  Text one  = new Text("1");//个数.      job.setCombinerClass(IndexCombiner.class);//优化. 上面这个代码可以处理单机节点的,加入是多台机器执行mapper函数,那么就会出现问题.

【原】MapReduce的排序和二次排序

- - ITeye博客
自己学习排序和二次排序的知识整理如下. 1.Hadoop的序列化格式介绍:Writable. 2.Hadoop的key排序逻辑. 4.如何自定义自己的Writable类型. 1.Hadoop的序列化格式介绍:Writable. 要了解和编写MR实现排序必须要知道的第一个知识点就是Writable相关的接口和类,这些是HADOOP自己的序列化格式.

利用hadoop mapreduce 做数据排序

- - zzm
我们的需求是想统计一个文件中用IK分词后每个词出现的次数,然后按照出现的次数降序排列. 由于hadoop在reduce之后就不能对结果做什么了,所以只能分为两个job完成,第一个job统计次数,第二个job对第一个job的结果排序. 第一个job的就是hadoop最简单的例子countwords,我要说的是用hadoop对结果排序.

Spark 颠覆 MapReduce 保持的排序记录

- - 开源中国社区最新新闻
在过去几年,Apache Spark的采用以惊人的速度增加着,通常被作为MapReduce后继,可以支撑数千节点规模的集群部署. 在内存中数 据处理上,Apache Spark比MapReduce更加高效已经得到广泛认识;但是当数据量远超内存容量时,我们也听到了一些机构在Spark使用 上的困扰. 因此,我们与Spark社区一起,投入了大量的精力做Spark稳定性、扩展性、性能等方面的提升.

堆排序

- kongshanzhanglao - 博客园-首页原创精华区
       堆排序是利用堆的性质进行的一种选择排序.   堆实际上是一棵完全二叉树,其任何一非叶节点满足性质:.   Key[i]<=key[2i+1]&&Key[i]<=key[2i+2]或者Key[i]>=Key[2i+1]&&key>=key[2i+2].   即任何一非叶节点的关键字不大于或者不小于其左右孩子节点的关键字.

排序算法

- - 互联网 - ITeye博客
排序算法有很多,所以在特定情景中使用哪一种算法很重要. 为了选择合适的算法,可以按照建议的顺序考虑以下标准: .     对于数据量较小的情形,(1)(2)差别不大,主要考虑(3);而对于数据量大的,(1)为首要.  一、冒泡(Bubble)排序——相邻交换 .  二、选择排序——每次最小/大排在相应的位置 .

lucene排序

- - 开源软件 - ITeye博客
排序是对于全文检索来言是一个必不可少的功能,在实际运用中,排序功能能在某些时候给我们带来很大的方便,比如在淘宝,京东等一些电商网站我们可能通过排序来快速找到价格最便宜的商品,或者通过排序来找到评论数最高或卖的最好的商品,再比如在Iteye里的博客栏里,每天都会以降序的方式,来显示出最新发出的几篇博客,有了排序,我们就能在某些时候很方便快速的得到某些有效信息,所以说排序功能,无处不在 ^_^.

Java排序算法:归并排序

- - zzm
 Java排序算法(九):归并排序. 归并排序(Merge)是将两个(或两个以上)有序表合并成一个新的有序表,即把待排序序列分为若干个子序列,每个子序列是有序的. 然后再把有序子序列合并为整体有序序列. 归 并排序是建立在归并操作上的一种有效的排序算法. 该算法是采用分治法(Divide and Conquer)的一个非常典型的应用.

排序大比武

- niko - C++博客-首页原创精华区
     摘要: 该比武只比速度,单线程测试随机正整数,不包括蜗牛排序,它弃权啦,哈哈. 操作系统:Windows XP Pro SP3,英文版编译器:g++4.5.2(-O3)CPU: Intel Core2 Q9500内存:DDR3普条 1066MHz, 4GB 插入排序、选择排序和冒泡排序最慢,时间复杂度为O(n2),希尔排序的速度依赖于使用的增量序列,堆排序、归并排序和改进的快速排序处于中等水平,时间复杂度...  阅读全文.