MapOutputBuffer理解的三重境界

作者：Lubin Liu

摘要

第一重：基本认识

1.1 MapOutputBuffer是什么？

1.3 为什么要理解这个类？

第二重：详细理解运行过程

一个metadata的长度是METASIZE=16
3. equator:缓冲区的中界点，raw的key value数据和meta信息分别从中界点往两侧填充。
4. kvindex:下次要插入的meta信息的起始位置
5. kvstart:溢写时meta数据的起始位置

6. kvend:溢写时meta数据的结束位置
7. bufindex:raw数据的结束位置
8. bufstart:溢写时raw数据的起始位置
9. bufend:溢写时raw数据的结束位置
10. spillper:当数据占用超过这个比例，会造成溢写，由配置“mapreduce.map.sort.spill.percent”指定，默认值是0.8
11. sortmb:kvbuffer占用的内存总量，单位是M，由配置“mapreduce.task.index.cache.limit.bytes”指定，默认值是100
12. indexCacheMemoryLimit:存放溢写文件信息的缓存大小，由参数“mapreduce.task.index.cache.limit.bytes”指定，单位是byte，默认值是1024*1024（1M)
13. bufferRemaining:buffer剩余空间，字节为单位
14. softLimit:字节单位的溢写阈值，超过之后需要写磁盘，值等于sortmb*spillper

2.3 初始状态

初始时，equator位于0处，在写入数据时，raw data往下标增大的方向延伸，meta data沿着下标减小的方向扩展，如果按照环形缓冲区来看待，raw data永远是顺时针扩展的，meta data是逆时针扩展的。各个量的初始化情况也很清楚，bufstart，bufend，bufindex都位于起点0处，kvstart, kvend,kvindex沿逆时针方向推进4个int的大小，因为他们标记的是meta data开始要放置的位置。另一个重要变量，标记剩余空间的bufferRemaining因为开始时buffer为空，所以其值等于softLimit。

2.4 第一组<key, value>对放入

2.5 第一次达到spill 阈值

2.6 溢写与更新

溢写开始之前，首先会更新kvend和bufend的位置，kvend等于kvindex+4int，因为每次kvindex总是标示下次要插入的位置，回退4int才是最后meta data写入的位置，kvstart不变，标志第一个metadata写入的位置。bufend是最后一条记录value结束的位置，bufstart标志第一个key开始的位置。kvstart与kvend之间是全部meta data，bufstart与bufend之间是raw data。溢写过程是守护线程完成的，在溢写过程中正常的写入缓冲区还可以继续。这也就是spillper的用处，1-spiller就近似代表了溢写触发时还可以使用的比例。
写入需要确定新的equator，新的equator一定位于kvend逆时针到bufend之间，也就是在空余空间的某个位置。我们期望写入过程尽可能的不要受溢写过程的影响，也就是写入的时候要利用当前的空余空间，而不应该侵入已经有数据的空间，新的equator要尽可能均匀的切分空闲空间，均匀的标准是要能放入meta+key+value组合。确定了equator就可以更新kvindex和bufindex的位置了，meta数据与raw数据的扩展方向是不变的，kvindex还是在下一个要插入meta的起始位置。bufferRemaining也会相应更新。剩下的可写入空间不应该是所有的空余空间，而是equator到kvend,bufend之间空间比较小的那个。任何一个用完，都代表不能继续写入了。

第三重：源码级详细分析

3.4 溢写线程的同步

3.4.1 同步控制综述

好在java1.5之后引入了“Condition”。此对象的await和signal方法对应于object的wait和notify。Condition由Lock创建，与Lock绑定，需要运行于“lock.lock"与”lock.unlock"之间。一个lock可以 有多个condition，就像在锁上有状态队列，可以精确控制唤醒哪个状态。ReentrantLock是Lock的一 个实现，spillThread正是由这种锁控制的。初始化时，有下列语句：

spillLock对象有两种Condition，spillDone和spillReady。需要同步的两个线程是主线程和spillThread。另外有两个boolean量表示状态，spillThreadRunning是volatile的，保证同时对两个线程看见。

3.4.2 spillDone

3.4.3 spillReady

3.4.4 spillThreadRunning

3.4.5 spillInProgress

3.5 溢写时数据排序的细节

private void sortAndSpill() throws IOException, ClassNotFoundException,
InterruptedException {
//approximate the length of the output file to be the length of the
//buffer + header lengths for the partitions
final long size = (bufend >=bufstart
? bufend -bufstart
: (bufvoid - bufend) + bufstart) +
partitions * APPROX_HEADER_LENGTH;
FSDataOutputStream out = null;
try {
// create spill file
final SpillRecord spillRec = new SpillRecord(partitions);
final Path filename =
mapOutputFile.getSpillFileForWrite(numSpills, size);
out = rfs.create(filename);
final int mstart = kvend / NMETA;
final int mend = 1 + // kvend is a valid record
(kvstart >=kvend
?kvstart
: kvmeta.capacity() + kvstart) / NMETA;
//对meta key进行排序，先按照partition，partition内部根据OutputKeyComparator
sorter.sort(MapOutputBuffer.this, mstart, mend, reporter);
int spindex = mstart;
final IndexRecord rec = new IndexRecord();
final InMemValBytes value = new InMemValBytes();
//循环遍历各个partition
for (int i = 0; i < partitions; ++i) {
IFile.Writer<K, V> writer = null;
try {
long segmentStart = out.getPos();
//对应此次循环的写入
writer = new Writer<K, V>(job, out, keyClass, valClass, codec,
spilledRecordsCounter);
if (combinerRunner == null) {
// spill directly
DataInputBuffer key = new DataInputBuffer();
while (spindex < mend &&
kvmeta.get(offsetFor(spindex % maxRec) + PARTITION) == i) {
final int kvoff = offsetFor(spindex % maxRec);
int keystart = kvmeta.get(kvoff + KEYSTART);
int valstart = kvmeta.get(kvoff + VALSTART);
key.reset(kvbuffer, keystart, valstart - keystart);
getVBytesForOffset(kvoff, value);
writer.append(key, value);
++spindex;
}
} else {
int spstart = spindex;
while (spindex < mend &&
kvmeta.get(offsetFor(spindex % maxRec)
+ PARTITION) == i) {
++spindex;
}
// Note: we would like to avoid the combiner if we've fewer
// than some threshold of records for a partition
if (spstart != spindex) {
combineCollector.setWriter(writer);
RawKeyValueIterator kvIter =
new MRResultIterator(spstart, spindex);
combinerRunner.combine(kvIter, combineCollector);
}
}
// close the writer
writer.close();
// record offsets
rec.startOffset = segmentStart;
rec.rawLength = writer.getRawLength();
rec.partLength = writer.getCompressedLength();
spillRec.putIndex(rec, i);
writer = null;
} finally {
if (null != writer) writer.close();
}
}
if (totalIndexCacheMemory >= indexCacheMemoryLimit) {
// create spill index file
Path indexFilename =
mapOutputFile.getSpillIndexFileForWrite(numSpills,partitions
* MAP_OUTPUT_INDEX_RECORD_LENGTH);
spillRec.writeToFile(indexFilename, job);
} else {//加入缓存
indexCacheList.add(spillRec);
totalIndexCacheMemory +=
spillRec.size() * MAP_OUTPUT_INDEX_RECORD_LENGTH;
}
LOG.info("Finished spill " + numSpills);
++numSpills;
} finally {
if (out != null) out.close();
}
}

3.6 收尾工作

spillThread.interrupt();
spillThread.join();

杀掉spillThread。
将小文件合并成大文件是在函数mergeParts中完成的，代码稍微有点长，但过程很清楚。每个mapper最终只输出一个数据文件finalOutputFile和一个与之对应的index文件finalIndexFile。我们知道，mapper时段用的partition，实际上与reducer对应，reducer会拉取属于自己的那段
partition，因此，finalOutputFile中，数据是按照partition来存放的，每个finalIndexFile文件中的项，就指向了finalOutputFile中的一个partition，存储了它的起始位置，压缩前长度和压缩后长度。回想一下，我们在输出每次spill的小文件时，也是按照partition存放的，并且也有一个index文件，因
此，最后合并的时候有一个二重循环，外层循环遍历partition，内层循环遍历spill file，每次把属于一个partition的那些小segment取出来，使用归并排序合并到一起，写入最终的文件，并记录index即与输出小文件时相同，最终这次合并，也可选的会调用combiner。

结语

参考文献
[1] MapOutputBuffer, http://grepcode.com/file/repo1.maven.org/maven2/org.apache.servicemix.bundles/org.apache.servicemix.bundles.hadoop-client/2.4.0_1/org/apache/hadoop/mapred/MapTask.java#MapTask.MapOutputBuffer