map写数据到本地磁盘过程解析----spill和merge

private final MapOutputCollector<K,V> collector;
private final org.apache.hadoop.mapreduce.Partitioner<K,V> partitioner;
private final int partitions;

private volatile int kvstart = 0; //表示当前已写的数据的开始位置
private volatile int kvend = 0; // 未执行spill是等于kvstart，执行spill是不等于kvstart（等于kvindex）
private int kvindex = 0; // 表示下一个可写的位置
private final int[] kvoffsets;
//volatile修改的字段表示此字段可以被不同线程访问和修改

private volatile int bufstart = 0; // marks beginning of spill
private volatile int bufend = 0; // marks beginning of collectable
private volatile int bufvoid = 0; // marks the point where we should stop
// reading at the end of the buffer
private int bufindex = 0; // marks end of collected
private int bufmark = 0; // marks end of record
private byte[] kvbuffer;

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
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 endPosition = (kvend > kvstart)
? kvend
: kvoffsets.length + kvend;
sorter.sort(MapOutputBuffer.this, kvstart, endPosition, reporter);
int spindex = kvstart;
IndexRecord rec = new IndexRecord();
InMemValBytes value = new InMemValBytes();
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 < endPosition &&
kvindices[kvoffsets[spindex % kvoffsets.length]
+ PARTITION] == i) {
final int kvoff = kvoffsets[spindex % kvoffsets.length];
getVBytesForOffset(kvoff, value);
key.reset(kvbuffer, kvindices[kvoff + KEYSTART],
(kvindices[kvoff + VALSTART] -
kvindices[kvoff + KEYSTART]));
writer.append(key, value);
++spindex;
}
} else {
int spstart = spindex;
while (spindex < endPosition &&
kvindices[kvoffsets[spindex % kvoffsets.length]
+ 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 >= INDEX_CACHE_MEMORY_LIMIT) {
// 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();
}
}

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 < endPosition &&
kvindices[kvoffsets[spindex % kvoffsets.length]
+ PARTITION] == i) {
final int kvoff = kvoffsets[spindex % kvoffsets.length];
getVBytesForOffset(kvoff, value);
key.reset(kvbuffer, kvindices[kvoff + KEYSTART],
(kvindices[kvoff + VALSTART] -
kvindices[kvoff + KEYSTART]));
writer.append(key, value);
++spindex;
}
} else {
int spstart = spindex;
while (spindex < endPosition &&
kvindices[kvoffsets[spindex % kvoffsets.length]
+ 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();
}
}

rec.startOffset = segmentStart;
43 rec.rawLength = writer.getRawLength();
44 rec.partLength = writer.getCompressedLength();
45 spillRec.putIndex(rec, i);

final SpillRecord spillRec = new SpillRecord(partitions);

IndexRecord rec = new IndexRecord();

startOffset；//数据的起始位置，也就是一个分区的键值对的起始位置
rawLength;//数据原始长度
partLength;//数据压缩后的长度(采用压缩技术)

if (totalIndexCacheMemory >= INDEX_CACHE_MEMORY_LIMIT) {
// 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;
}

private static final int INDEX_CACHE_MEMORY_LIMIT = 1024 * 1024;
private int totalIndexCacheMemory;

private ArrayList<SpillRecord> indexCacheList;
private int MAP_OUTPUT_INDEX_RECORD_LENGTH=24;

totalIndexCacheMemory += spillRec.size() * MAP_OUTPUT_INDEX_RECORD_LENGTH;

io.sort.mb=;//用于存储map输出键值对的缓存，默认值是100(100M)
io.sort.spill.percent=0.1;//缓存spill的比例，默认是0.8.
//将以上两个参数修改之后就可以使map更多的spill数据到硬盘上，一个直观的效果就是spill的文件有很多。我们知道IO是很耗时间的，相当于“拖慢”了map的速度。
//另外，还有一个参数:
io.sort.factor=;//这个参数默认是10，表示执行合并spill文件的时候最多一次可以合并多少个spill文件，调下这个值可以看到合并的过程，也就是看到file.out（所有spill文件合并后的大文件）和spill.out并存

hadoop.tmp.dir="/home/hadoop/HadoopData/tmp/";//这是hadoop的临时基础目录，很对目录都依赖它
mapred.local.dir=${hadoop.tmp.dir}/mapred/local;//这是默认值，也就是mapred的中间数据文件的存储目录。