现在的位置: 首页 > 综合 > 正文

Hadoop系列 之Terasort

2014年09月05日 ⁄ 综合 ⁄ 共 6756字 ⁄ 字号 评论关闭

TeraSort是Hadoop的测试中很有用的一个工具,但以前只是粗略的知道它的功能和用法,简单的用它做了几个测试用例。实际上,对于这种比较通用的工具,如果能够了解它更多一些的话,对于理解Hadoop是很有帮助的,同时也可以更好的利用它来帮助测试。最近有点时间,就了解了一些它的背景,代码实现原理等等,就先记录下来吧。

1. Hadoop与Sort Benchmarks

SortBenchmark(http://sortbenchmark.org/ )是JimGray自98年建立的一项排序竞技活动,它制定了不同类别的排序项目和场景,每年一次,决出各项排序算法实现的第一名(看介绍是在每年的ACM
SIGMOD颁发奖牌哦)。

Hadoop在2008年以209秒的成绩获得年度TeraSort项(Dotona类)的第一名;而此前这一项排序的记录是297秒。

从SortBenchmark网站上可以了解到,Hadoop到今天仍然保持了Minute项Daytona类型排序的冠军。Minute项排序是通过评判在60秒或小于60秒内能够排序的最大数据量来决定胜负的;其实等同于之前的TeraSort(TeraSort的评判标准是对1T数据排序的时间)。

Hadoop源代码中包含了TeraSort,打包在examples包(如:hadoop-0.20.2-examples.jar)。

2. 输入数据:TeraGen

SortBenchmark对排序的输入数据制定了详细规则,要求使用其提供的gensort工具(http://www.ordinal.com/gensort.html )生成输入数据。Hadoop的TeraSort也用Java实现了一个生成数据工具TeraGen,算法与gensort一致。

对输入数据的基础要求是:输入文件是由一行行100字节的记录组成,每行记录包括一个10字节的Key;以Key来对记录排序。

Minute项排序允许输入文件可以是多个文件,但Key的每个字节要求是binary编码而不是ASCII编码,也就是每个字符可能有256种可能,也就是说每条记录,有2的80次方种可能的Key;

同时Daytona类别则要求排序程序不仅是为10字节长Key、100字节长记录排序设计的,还可以支持对其他长度的Key或行记录进行排序;也就是说这个排序程序是通用的。

在hadoop里,利用TeraGen生成排序输入数据的命令格式是这样的:

[html] view
plain
copy

  1. $ bin/hadoop jar hadoop-0.19.2-examples.jar teragen 10000000000 /terasort/input1TB  

注意,teragen后的数值单位是行数;因为每行100个字节,所以如果要产生1T的数据量,则这个数值应为1T/100=10000000000(10个0)。

 

生成的数据是这样的:

[html] view
plain
copy

  1. !x'-n[Pp+l1049085170QQQQQQQQQQRRRRRRRRRRSSSSSSSSSSTTTTTTTTTTUUUUUUUUUUVVVVVVVVVVWWWWWWWWWWXXXXXXXX  
  2. r0JZ8-|o\)1049085171YYYYYYYYYYZZZZZZZZZZAAAAAAAAAABBBBBBBBBBCCCCCCCCCCDDDDDDDDDDEEEEEEEEEEFFFFFFFF  
  3. @Jp9XC#d/J1049085172GGGGGGGGGGHHHHHHHHHHIIIIIIIIIIJJJJJJJJJJKKKKKKKKKKLLLLLLLLLLMMMMMMMMMMNNNNNNNN  
  4. N)eM''3<pr1049085173OOOOOOOOOOPPPPPPPPPPQQQQQQQQQQRRRRRRRRRRSSSSSSSSSSTTTTTTTTTTUUUUUUUUUUVVVVVVVV  
  5. ryfUS$G1&y1049085174WWWWWWWWWWXXXXXXXXXXYYYYYYYYYYZZZZZZZZZZAAAAAAAAAABBBBBBBBBBCCCCCCCCCCDDDDDDDD  
  6. =i*nyMblSg1049085175EEEEEEEEEEFFFFFFFFFFGGGGGGGGGGHHHHHHHHHHIIIIIIIIIIJJJJJJJJJJKKKKKKKKKKLLLLLLLL  
  7. 7I6>/,!~@@1049085176MMMMMMMMMMNNNNNNNNNNOOOOOOOOOOPPPPPPPPPPQQQQQQQQQQRRRRRRRRRRSSSSSSSSSSTTTTTTTT  
  8. #g{6{0Z;%\1049085177UUUUUUUUUUVVVVVVVVVVWWWWWWWWWWXXXXXXXXXXYYYYYYYYYYZZZZZZZZZZAAAAAAAAAABBBBBBBB  
  9. 7</ioXV\It1049085178CCCCCCCCCCDDDDDDDDDDEEEEEEEEEEFFFFFFFFFFGGGGGGGGGGHHHHHHHHHHIIIIIIIIIIJJJJJJJJ  
  10. 8in+{B{w'R1049085179KKKKKKKKKKLLLLLLLLLLMMMMMMMMMMNNNNNNNNNNOOOOOOOOOOPPPPPPPPPPQQQQQQQQQQRRRRRRRR  
  11. ]Xq/CdFy%E1049085180SSSSSSSSSSTTTTTTTTTTUUUUUUUUUUVVVVVVVVVVWWWWWWWWWWXXXXXXXXXXYYYYYYYYYYZZZZZZZZ  
  12. :,/$4U]DIJ1049085181AAAAAAAAAABBBBBBBBBBCCCCCCCCCCDDDDDDDDDDEEEEEEEEEEFFFFFFFFFFGGGGGGGGGGHHHHHHHH  

每行记录由3段组成:

  • 前10个字节:随机binary code的十个字符,为key
  • 中间10个字节:行id
  • 后面80个字节:8段,每段10字节相同随机大写字母

在这我其实有个小问题,因为排序是针对Key的,岂不是后面的Value只需要保证长度就可以,至于内容是什么都没关系么?有人可能说随机字母组成是为了避免压缩时不均衡。但SortBenchmark是要求所有输入文件、输出文件、甚至中间传输过程的文件都不允许使用压缩的。不管怎么说,对内容制定个规则,应该还是为了保证比赛公平性吧。

TeraGen作业没有Reduce Task,产生文件的个数取决于设定Map的个数。


3. TeraSort的原理

了解了我们需要排序的输入数据是什么样了,接下来来了解TeraSort。


运行TeraSort的命令是这样的: 

[java] view
plain
copy

  1. $ bin/hadoop jar hadoop-0.19.2-examples.jar terasort /terasort/input1TB /terasort/output1TB  

运行后,我们可以看到会起m个mapper(取决于输入文件个数)和r个reducer(取决于设置项:mapred.reduce.tasks),排好序的结果存放在/terasort/output1TB目录。

查看TeraSort的源代码,你会发现这个作业同时没有设置mapper和reducer;也就是意味着它使用了Hadoop默认的IdentityMapper和IdentityReducer。IdentityMapper和IdentityReducer对它们的输入不做任何处理,将输入k,v直接输出;也就是说是完全是为了走框架的流程而空跑。
 
这正是Hadoop的TeraSort的巧妙所在,它没有为排序而实现自己的mapper和reducer,而是完全利用Hadoop的Map Reduce框架内的机制实现了排序。
 
我们需要先来了解下Hadoop的Map Reduce过程,下图较清楚的说明了这个过程。

(这个图非常经典,截自:Hadoop:The Definitive Guide)
 
恰好趁这个机会,把自己对MapReduce过程的理解简要的整理一下:

  • 一般情况下,作业会需要指定input目录和output目录
  • 作业的Mapper根据设置的InputFormat来从input目录读取输入数据,分成多个splits; 每一个split交给一个mapper处理
  • mapper的输出会按照partions分组,每一个partion对应着一个reducer的输入; 在每个partion内,会有一个按key排序的过程,也就是说,每一个partion内的数据是有序的
  • 当处理完combiner和压缩后(如果有设置),map的输出会写到硬盘上。map结束后,所在的TT会在下一个心跳通知到JT。
  • 每一个reducer查询JT了解到属于自己对应partition的mapoutput数据的对应的TT位置,然后去那copy到本地(HTTP协议)。
  • Copy并保存到本地磁盘的过程同mapper端的输出保存过程非常相似。等到reducer获取到属于它的所有mapoutput数据后,它会保持之前mapper端的sort顺序,把这些mapoutput合并成较集中的中间文件(个数取决于数据大小和设置)。为了节省io的开销,merge会保证最后一轮是满负荷合并;并且,merge的最后一轮输出会直接在内存输入给reducer。
  • reducer的输出按照OutputFormat来保存到output目录。

 
如果我们注意到上述过程的蓝色加粗部分,就可以猜测到TeraSort是如何利用hadoop的map reduce机制来达到排序目的的。或许我们可以怀疑,hadoop的Map Reduce机制就是为了TeraSort而设计的,:) 。

 
既然Hadoop可以保证每一个partition内的数据有序,TeraSort只需要做一件事情:保证partition之间也是有序的。

Hadoop默认的partitioner是HashPartitioner,它的实现是这样的:

[java] view
plain
copy

  1. public classHashPartitioner<K2, V2> implements Partitioner<K2, V2> {  
  2.   public void configure(JobConf job) {}  
  3.   public int getPartition(K2 key, V2 value,  
  4.                           int numPartitions) {  
  5.     return (key.hashCode() &Integer.MAX_VALUE) % numPartitions;  
  6.   }  
  7. }  


它将每个Key的HashCode对总reducer数取模,转换成partion index。
个人理解这样做有两个目的:

  • 所有相同Key的数据在一个Reducer内处理
  • 尽量均匀的将数据分配到各个Reducer

但毫无疑问,HashPartitioner不能保证它的Partion之间的有序。
 
为了保证Partion之间的有序,TeraSort定义了一个TotalOrderPartitioner。

TotalOrderPartitioner首先要解决的问题是,partitioner发生在map里,而每个mapper只处理它自己的一份split数据,它如何知道它所处理的数据在全局所有输入数据里的位置?
 
回溯到源头,InputFormat有数据的全局观。TeraSort定义的TeraInputFormat有一个重要的功能,就是把对全部数据形成一个摘要文件,以提供给之后的partitioner使用。为了保证保证效率,TeraSort采用抽样来实现摘要。

运行TeraSort后,它做的第一件事情是对输入数据进行抽样,抽样频率由设置项抽样总条数terasort.partitions.sample决定,默认值为100000条。对输入记录分10个区间(或更小)来分批采样,直到采到足够条数;采样完成后对这些抽样点进行排序,然后对排序后记录均分成partitions个区间,最终将这些区间分割点写到文件,文件名为_partition.lst。这个文件会被加到distributed cache里,目的是为了能被hadoop分发到将来运行mapper的每一个TT上去。

有了所有数据的摘要信息,后面的partitioner做起来就有依据了。当它处理一条mapper的输出记录时,它可以按照一种映射算法,依据每条记录的Key与_partition.lst记录的对应信息做比较,将它划分到某一个partition,从而保证partition之间的有序性。

假设我们从_partition.lst得到的Key组合为sample[];某一条记录的Key值为key,如果查找到sample[i-1] <= key < sample[i] , 那么这条记录会被分配到第i个partition,也就是第i个reducer来处理。这样,partition之间也就有序了。

在TeraSort中,构建了一个trie来实现对Key的查找归类(Partitioner的过程其实就是归类,然后把每一类交给一个reducer处理)。TeraSort默认使用2层的Trie,意味着它只用Key的前两个字节与与分割点比较; Trie的非叶子节点有256个子节点(对应着Key的每一个字节的binary code,有256种可能)。

需要提到的是,TeraSort输出的replica数设置是1份,而不是Hadoop默认使用的3份。为什么?因为SortBenchmark没有规定结果要存多份副本,而设置成1份,Hadoop会就近存在本地(如果这个reducer的TT上也同时有DN)。这可节省了不少网络和磁盘消耗,间接的提高了TeraSort的执行效率。

4. 结果的校验:TeraValidate

TeraSort还带有一个校验程序,来检验排序输出结果是否是有序的。TeraValidate是一个简单的Map Reduce作业,大家看看Mapper和Reducer就好了。
执行TeraValidate的命令是:

[java] view
plain
copy

  1. bin/hadoop jar hadoop-0.19.2-examples.jar teravalidate /terasort/output1TB /terasort/validate1TB  


如果有错误,log记录会放在输出目录里。


需要一提的是TeraValidate的作业配置里有这么一句:

[java] view
plain
copy

  1. job.setLong("mapred.min.split.size", Long.MAX_VALUE);  


它用来保证每一个输入文件都不会被split,又因为TeraInputFormat继承自FileInputFormat,所以TeraValidate运行mapper的总数正好等于输入文件的个数。


5. TeraSort与hadoop测试

TeraSort巧妙的利用了Hadoop的MapReduce机制来实现了Sort的目的,与Hadoop机制的完美结合也许是它优异排序成绩的一个重要原因。而也正因为如此,我们可以在集群上利用TeraSort来测试Hadoop,它将具有很高的测试利用价值。

我稍微总结下,可以用TeraSort来测试的场景:

  • 在不同版本Hadoop上运行TeraSort,使用相同的配置参数,来进行正确性对比测试
  • 在不同版本Hadoop上运行TeraSort,使用相同的配置参数,来进行性能对比测试
  • 在相同版本Hadoop上运行TeraSort,使用不同的配置参数,进行性能对比测试,发现问题

尤其是第三项,可以衍生许许多多的测试用例;也可以利用第三项来作为Hadoop作业调优的依据。

TeraSort只是一个小工具,比起生产应用作业,可能是微不足道了。但一个小工具,如果能够挖掘到底,背后也会有大价值;尤其对测试来讲,如果能够对背景知识有更多的了解,一个小工具可以转换成众多方便且有价值的测试用例;并且,如果能对一个小工具举一反三,也能够为其他地方的测试提供价值。


ZZ from http://blog.csdn.net/leafy1980/article/details/6633828

Thanks to leafy1980


抱歉!评论已关闭.