环形缓冲区底层实现
首先明白改过程发生在Map——Collect阶段:在用户编写的map() 函数中,当数据处理完成后,一般会调用OutputCollector.collect() 输出结果。在该函数内部,它会将生成的key/value分片(通过调用Partitioner),并写入一个环形内存缓冲区中。 MapOutputBuffer 内部使用了一个缓冲区暂时存储用户输出数据,当缓冲区使用率达到一定阈值后,再将缓冲区中的数据写到磁盘上。
数据缓冲区的设计方式直接影响到Map Task的写效率,而现有多种数据结构可供选择,最简单的是单向缓冲区,生产者向缓冲区中单向写入输出,当缓冲区写满后,一次性写到磁盘上,就这样,不断写缓冲区,直到所有数据写到磁盘上。单向缓冲区最大的问题是性能不高,不能支持同时读写数据。
双缓冲区是对单向缓冲区的一个改进,它使用两个缓冲区,其中一个用于写入数据,另一个将写满的数据 写到磁盘上,这样,两个缓冲区交替读写,进而提高效率。实际上,双缓冲区只能一定程度上让读写并行,仍会存在读写等待问题。
一种更好的缓冲区设计方式是采用环形缓冲区:当缓冲区使用率达到一定阈值后,便开始向磁盘上写入数据,同时,生产者仍可以向不断增加的剩余空间中循环写入数据,进而达到真正的读写并行。
底层就是一个字节数组:数组前面记录关于KV的索引位置,数组后面记录KV数据。首尾相接构成一个环形的缓冲区,中间是赤道。用于数据spll溢出处理。
单生产者消费者模型,其中,MapOutputBuffer的collect 方法和MapOutputBuffer.Buffer 的write 方法是生 产者,spillThread 线程是消费者,它们之间同步是通过可重入的互斥锁spillLock 和spillLock 上的两个条件变量(spillDone 和spillReady )完成 的.生产者主要代码如下
spillLock.lock();
if(缓冲区使用率达到阈值){
spillReady.signal();
}
if(缓冲区满){
spillDone.wait();
}
spillLock.lock();
MapOutputBuffer 内部采用了两级索引结构,涉及三个环形内存缓冲区,分别是kvoffsets 、kvindices 和kvbuffer ,这三个缓冲 区所占内存空间总大小为io.sort.mb (默认是100 MB)。
kvoffsets即偏移量索引数组,用于保存key/value 信息在位置索引kvindices 中的偏移量。考虑到一对key/value 需占用数组kvoffsets 的1个 int (整型)大小,数组kvindices 的3个int大小(分别保存所在partition号、key开始位置和value开始位置),所以Hadoop按比例1:3将大小为 ${io.sort.record.percent}*${io.sort.mb} 的内存空间分配给数组kvoffsets 和kvindices ,
该过 程由指针kvstart/kvend/kvindex控制,其中kvstart表示存有数据的内存段初始位置,kvindex表示未存储数据的内存段初始位置,而在正常写 入情况下,kvend=kvstart,一旦满足溢写条件,则kvend=kvindex,此时指针区间[kvstart, kvend)为有效数据区间。
kvindices即位置索引数组,用于保存key/value值在数据缓冲区kvbuffer中的起始位置。
kvbuffer即数据缓冲区,用于保存实际的key/value值,默认情况下最多可使用io.sort.mb 中的95%,当该缓冲区使用率超过 io.sort.spill.percent (默认80%)后,便会触发线程SpillThread将数据写入磁盘。
更多的详细信息你可以参考《Hadoop技术内幕:深入解析MapReduce架构设计与实现原理 》 你可以点这里扫描二维码(或者微信搜 :孙中明) 回复关键字 3006 获取相关书籍此类
|