码迷,mamicode.com
首页 > 其他好文 > 详细

Hadoop源代码分析(MapTask辅助类,II)

时间:2015-06-02 17:57:00      阅读:112      评论:0      收藏:0      [点我收藏+]

标签:hadoop   超人学院   

有了上面Mapper输出的内存存储结构和硬盘存储结构讨论,我们来仔细分析MapOutputBuffer的流程。首先是成员变量。最先初始化的是作业配置job和统计功能reporter。通过配置,MapOutputBuffer可以获取本地文件系统(localFs和rfs),Reducer的数目和Partitioner。
SpillRecord是文件spill.out{spill号}.index在内存中的对应抽象(内存数据和文件数据就差最后的校验和),该文件保持了一系列的IndexRecord,如下图:

技术分享 
IndexRecord有3个字段,分别是startOffset:记录偏移量,rawLength:初始长度,partLength:实际长度(可能有压缩)。SpillRecord保持了一系列的IndexRecord,并提供方法用于添加记录(没有删除记录的操作,因为不需要),获取记录,写文件,读文件(通过构造函数)。
接下来是一些和输出缓存区kvbuffer,缓存区记录索引kvindices和缓存区记录索引排序工作数组kvoffsets相关的处理,下面的图有助于说明这段代码。

技术分享 

这部分依赖于3个配置参数,io.sort.spill.percent是kvbuffer,kvindices和kvoffsets的总大小(以M为单位,缺省是100,就是100M,这一部分是MapOutputBuffer中占用存储最多的)。io.sort.record.percent是kvindices和kvoffsets占用的空间比例(缺省是0.05)。前面的分析我们已经知道kvindices和kvoffsets,如果记录数是N的话,它占用的空间是4N*4bytes,根据这个关系和io.sort.record.percent的值,我们可以计算出kvindices和kvoffsets最多能有多少个记录,并分配相应的空间。参数io.sort.spill.percent指示当输出缓冲区或kvindices和kvoffsets记录数量到达对应的占用率的时候,会启动spill,将内存缓冲区的记录存放到硬盘上,softBufferLimit和softRecordLimit为对应的字节数。
值对<key, value>输出到缓冲区是通过Serializer串行化的,这部分的初始化跟在上面输出缓存后面。接下来是一些计数器和可能的数据压缩处理器的初始化,可能的Combiner和combiner工作的一些配置。
最后是启动spillThread,该Thread会检查内存中的输出缓存区,在满足一定条件的时候将缓冲区中的内容spill到硬盘上。这是一个标准的生产者-消费者模型,MapTask的collect方法是生产者,spillThread是消费者,它们之间同步是通过spillLock(ReentrantLock)和spillLock上的两个条件变量(spillDone和spillReady)完成的。
先看生产者,MapOutputBuffer.collect的主要流程是:
l           报告进度和参数检测(<K,V>符合Mapper的输出约定);
l           spillLock.lock(),进入临界区;
l           如果达到spill条件,设置变量并通过spillReady.signal(),通知spillThread;并等待spill结束(通过spillDone.await()等待);
l           spillLock.unlock();
l           输出key,value并更新kvindices和kvoffsets(注意,方法collect是synchronized,key和value各自输出,它们也会占用连续的输出缓冲区);
kvstart,kvend和kvindex三个变量在判断是否需要spill和spill是否结束的过程中很重要,kvstart是有效记录开始的下标,kvindex是下一个可做记录的位置,kvend的作用比较特殊,它在一般情况下kvstart==kvend,但开始spill的时候它会被赋值为kvindex的值,spill结束时,它的值会被赋给kvstart,这时候kvstart==kvend。这就是说,如果kvstart不等于kvend,系统正在spill,否则,kvstart==kvend,系统处于普通工作状态。其实在代码中,我们可以看到很多kvstart==kvend的判断。
下面我们分情况,讨论kvstart,kvend和kvindex的配合。初始化的时候,它们都被赋值0。

技术分享 

下图给出了一个没有spill的记录添加过程:

技术分享 

注意kvindex和kvnext的关系,取模实现了循环缓冲区
如果在添加记录的过程中,出现spill(多种条件),那么,主要的过程如下:

技术分享 

首先还是计算kvnext,主要,这个时候kvend==kvstart(图中没有画出来)。如果spill条件满足,那么,kvindex的值会赋给kvend(这是kvend不等于kvstart),从kvstart和kvend的大小关系,我们可以知道记录位于数组的那一部分(左边是kvstart<kvend的情况,右边是另外的情况)。Spill结束的时候,kvend值会被赋给kvstart,kvend==kvstart又重新满足,同时,我们可以发现kvindex在这个过程中没有变化,新的记录还是写在kvindex指向的位置,然后,kvindex=kvnect,kvindex移到下一个可用位置。
大家体会一下上面的过程,特别是kvstart,kvend和kvindex的配合,其实,<key,value>对输出使用的缓冲区,也有类似的过程。

Collect在处理<key,value>输出时,会处理一个MapBufferTooSmallException,这是value的串行化结果太大,不能一次放入缓冲区的指示,这种情况下我们需要调用spillSingleRecord,特殊处理。

更多精彩内容请关注:http://bbs.superwu.cn 

关注超人学院微信二维码:技术分享

关注超人学院java免费学习交流群:技术分享

Hadoop源代码分析(MapTask辅助类,II)

标签:hadoop   超人学院   

原文地址:http://blog.csdn.net/crxy2014/article/details/46332023

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!