加入收藏 | 设为首页 | 会员中心 | 我要投稿 | RSS
您当前的位置:首页 > 教程文章 > NOSQL数据库

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

时间:2012-04-27 01:05:55  来源:  作者:

接下来讨论的是key,value的输出,这部分比较复杂,不过有了前面kvstart,kvend和kvindex配合的分析,有利于我们理解这部分的代码。

输出缓冲区中,和kvstart,kvend和kvindex对应的是bufstart,bufend和bufmark。这部分还涉及到变量bufvoid,用于表明实际使用的缓冲区结尾(见后面BlockingBuffer.reset分析),和变量bufmark,用于标记记录的结尾。这部分代码需要bufmark,是因为key或value的输出是变长的,(前面元信息记录大小是常量,就不需要这样的变量)。

最好的情况是缓冲区没有翻转和value串行化结果很小,如下图:



 

 

先对key串行化,然后对value做串行化,临时变量keystart,valstart和valend分别记录了key结果的开始位置,value结果的开始位置和value结果的结束位置。

串行化过程中,往缓冲区写是最终调用了Buffer.write方法,我们后面再分析。

如果key串行化后出现bufindex < keystart,那么会调用BlockingBuffer的reset方法。原因是在spill的过程中需要对<key,value>排序,这种情况下,传递给RawComparator的必须是连续的二进制缓冲区,通过BlockingBuffer.reset方法,解决这个问题。下图解释了如何解决这个问题:

 

 

 



 

当发现key的串行化结果出现不连续的情况时,我们会把bufvoid设置为bufmark,见缓冲区开始部分往后挪,然后将原来位于bufmark到bufvoid出的结果,拷到缓冲区开始处,这样的话,key串行化的结果就连续存放在缓冲区的最开始处。

上面的调整有一个条件,就是bufstart前面的缓冲区能够放下整个key串行化的结果,如果不能,处理的方式是将bufindex置0,然后调用BlockingBuffer内部的out的write方法直接输出,这实际调用了Buffer.write方法,会启动spill过程,最终我们会成功写入key串行化的结果。

下面我们看write方法。key,value串行化过程中,往缓冲区写数据是最终调用了Buffer.write方法,又是一个复杂的方法。

l do-while循环,直到我们有足够的空间可以写数据(包括缓冲区和kvindices和kvoffsets)

u 首先我们计算缓冲区连续写是否写满标志buffull和缓冲区非连续情况下有足够写空间标志wrap(这个实在拗口),见下面的讨论;条件(buffull && !wrap)用于判断目前有没有足够的写空间;

u spill没启动的情况下(kvstart == kvend),分两种情况,如果数组中有记录(kvend != kvindex),那么,根据需要(目前输出空间不足或记录数达到spill条件)启动spill过程;否则,如果空间还是不够(buffull && !wrap),表明这个记录非常大,以至于我们的内存缓冲区不能容下这么大的数据量,抛MapBufferTooSmallException异常;

u 如果空间不足同时spill在运行,等待spillDone;

l 写数据,注意,如果buffull,则写数据会不连续,则写满剩余缓冲区,然后设置bufindex=0,并从bufindex处接着写。否则,就是从bufindex处开始写。

下图给出了缓冲区连续写是否写满标志buffull和缓冲区非连续情况下有足够写空间标志wrap计算的几种可能:

 



 

情况1和情况2中,buffull判断为从bufindex到bufvoid是否有足够的空间容纳写的内容,wrap是图中白颜色部分的空间是否比输入大,如果是,wrap为true;情况3和情况4中,buffull判断bufindex到bufstart的空间是否满足条件,而wrap肯定是false。明显,条件(buffull && !wrap)满足时,目前的空间不够一次写。

接下来我们来看spillSingleRecord,只是用于写放不进内存缓冲区的<key,value>对。过程很流水,首先是创建SpillRecord记录,输出文件和IndexRecord记录,然后循环,构造SpillRecord并在恰当的时候输出记录(如下图),最后输出spill{n}.index文件。



 

前面我们提过spillThread,在这个系统中它是消费者,这个消费者相当简单,需要spill时调用函数sortAndSpill,进行spill。sortAndSpill和spillSingleRecord类似,函数的开始也是创建SpillRecord记录,输出文件和IndexRecord记录,然后,需要在kvoffsets上做排序,排完序后顺序访问kvoffsets,也就是按partition顺序访问记录。

partition循环处理排完序的数组,如果没有combiner,则直接输出记录,否则,调用combineAndSpill,先做combin然后输出。循环的最后记录IndexRecord到SpillRecord。

sortAndSpill最后是输出spill{n}.index文件。

combineAndSpill比价简单,我们就不分析了。

BlockingBuffer中最后要分析的方法是flush方法。调用flush方法,意味着Mapper的结果都已经collect了,需要对缓冲区做一些最后的清理,并合并spill{n}文件产生最后的输出。

缓冲区处理部分很简单,先等待可能的spill过程完成,然后判断缓冲区是否为空,如果不是,则调用sortAndSpill,做最后的spill,然后结束spill线程。

flush合并spill{n}文件是通过mergeParts方法。如果Mapper最后只有一个spill{n}文件,简单修改该文件的文件名就可以。如果Mapper没有任何输出,那么我们需要创建哑输出(dummy files)。如果spill{n}文件多于1个,那么按partition循环处理所有文件,将处于处理partition的记录输出。处理partition的过程中可能还会再次调用combineAndSpill,最记录再做一次combination,其中还涉及到工具类Merger,我们就不再深入研究了。

来顶一下
返回首页
返回首页
发表评论 共有条评论
用户名: 密码:
验证码: 匿名发表
推荐资讯
在CentOS下搭建Android 开发环境
在CentOS下搭建Androi
轻松搭建属于自己的Ubuntu发行版
轻松搭建属于自己的Ub
利用SUSE Studio 打造自己的个性化Linux发行版
利用SUSE Studio 打造
那些采用PHP技术的IT大企业
那些采用PHP技术的IT大
相关文章
栏目更新
栏目热门