Hadoop源代码分析（MapTask辅助类，II），hadoopmaptask


有了上面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免费学习交流群：

相关内容