首页 > 代码库 > Hadoop-2.4.1学习之Map任务源码分析(下)

Hadoop-2.4.1学习之Map任务源码分析(下)

      在Map任务源码分析(上)中,对MAP阶段的代码进行了学习,这篇文章文章将学习Map任务的SORT阶段。如果Reducer的数量不为0,则还需要进行SORT阶段,但从上面的学习中并未发现与MAP阶段执行完毕调用mapPhase.complete()类似的在SORT阶段执行完毕调用sortPhase.complete()的源码,那SORT阶段是在什么时候启动的?对于Map任务来说,有输入就有输出,输入由RecordReader负责,输出则由RecordWriter负责,当Reducer的数量不为0时,RecordWriter为NewOutputCollector(该类为MapTask的私有内部类),SORT阶段对map的输出进行处理,由此推断SORT阶段的工作是由NewOutputCollector完成的,下面将通过分析NewOutputCollector的源代码要验证这一推断是否成立。该类继承自RecordWriter,拥有的变量如下:

private final MapOutputCollector<K,V> collector;//负责实际的输出操作
private final org.apache.hadoop.mapreduce.Partitioner<K,V> partitioner;//对键空间进行分区
private final int partitions;//分区数量,与Reducer的数量相同
      该类的构造函数如下:

collector = createSortingCollector(job, reporter);
partitions = jobContext.getNumReduceTasks();
if (partitions > 1) {
partitioner = (org.apache.hadoop.mapreduce.Partitioner<K,V>)
ReflectionUtils.newInstance(jobContext.getPartitionerClass(), job);
} else {
partitioner = new org.apache.hadoop.mapreduce.Partitioner<K,V>() {
@Override
public int getPartition(K key, V value, int numPartitions) {
return partitions - 1;
}
};
}

      在该段代码中还是未发现与SORT阶段有关的任何信息,但却发现了Sorting,据此推断方法createSortingCollector具有最大的可能性。该方法的源代码如下:

//根据mapreduce.job.map.output.collector.class的值构建MapOutputCollector
//在未指定该参数值的情况,返回MapOutputBuffer对象
MapOutputCollector<KEY, VALUE> collector= (MapOutputCollector<KEY, VALUE>)
ReflectionUtils.newInstance.
(job.getClass(JobContext.MAP_OUTPUT_COLLECTOR_CLASS_ATTR,
MapOutputBuffer.class, MapOutputCollector.class), job);
LOG.info("Map output collector class = " + collector.getClass().getName());
MapOutputCollector.Context context =new MapOutputCollector.Context(this, job, reporter);
//默认调用MapOutputBuffer的init方法
collector.init(context);
return collector;

      经过上面的一系列分析可知,SORT阶段的工作由NewOutputCollector完成,而NewOutputCollector又将SORT工作交给了MapOutputCollector,最终由该接口的实现类MapOutputBuffer完成,该类做为MapTask的内部类占用了MapTask源代码中超过一半的行数(MapTask行数为2000行,MapOutputBuffer约为1100行),但从行数就可以得出该类的重要性。MapOutputBuffer的init方法的第一部分代码根据设置构建缓存,代码已经添加了相关注释:

//sanity checks
//当kvbuffer缓存达到该值时,溢出线程将缓存内容写入硬盘
final float spillper =job.getFloat(JobContext.MAP_SORT_SPILL_PERCENT, (float)0.8);
//用于排序文件的缓存的大小,默认为100MB
final int sortmb = job.getInt(JobContext.IO_SORT_MB, 100);
//mapreduce.task.index.cache.limit.bytes,默认值为1024 * 1024(1M)
indexCacheMemoryLimit = job.getInt(JobContext.INDEX_CACHE_MEMORY_LIMIT,                                     INDEX_CACHE_MEMORY_LIMIT_DEFAULT);
if (spillper > (float)1.0 || spillper <= (float)0.0) {
    throw new IOException("Invalid \"" + JobContext.MAP_SORT_SPILL_PERCENT +
     "\": " + spillper);
}
//sortmb的最大值为2047Mb(111 1111 1111),取sortmb的最低11位
//若大于2047Mb,下面的左移20位将导致溢出
if ((sortmb & 0x7FF) != sortmb) {
    throw new IOException("Invalid \"" + JobContext.IO_SORT_MB + "\": " + sortmb);
}
//默认使用快速排序
sorter = ReflectionUtils.newInstance(job.getClass("map.sort.class",
QuickSort.class, IndexedSorter.class), job);
// buffers and accounting
//sortbm*(2的20次方),将sortbm转换为字节数(1024*1024,2的十次方乘以2的十次方)
int maxMemUsage = sortmb << 20;
//METASIZE=16,maxMemUsage=sortmb << 20
maxMemUsage -= maxMemUsage % METASIZE;
kvbuffer = new byte[maxMemUsage];
bufvoid = kvbuffer.length;
kvmeta = ByteBuffer.wrap(kvbuffer).order(ByteOrder.nativeOrder()).asIntBuffer();
setEquator(0);
//equator:标记元数据或者序列化数据的起源
//bufstart:标记溢出的起始位置,bufend:标记可收集收据的起始位置
//bufindex:标记已收集数据的结束位置。全部初始化为0
bufstart = bufend = bufindex = equator;
//kvstart:标记溢出元数据的起源,kvend:标记溢出元数据的结束位置
//kvindex:标记完全序列化的记录的结束位置
kvstart = kvend = kvindex;
maxRec = kvmeta.capacity() / NMETA;
softLimit = (int)(kvbuffer.length * spillper);
bufferRemaining = softLimit;

      MapOutputBuffer的init方法的第二部分启动SpillThread线程,该线程用于完成SORT阶段的工作,并负责溢出缓存中的数据。在该线程中的run方法中调用了sortAndSpill方法,由方法名就可以得知该方法负责map输出的排序和溢出工作,排序部分的源代码如下:

final int mstart = kvend / NMETA;
// kvend is a valid record
final int mend = 1 + (kvstart >= kvend? kvstart
          : kvmeta.capacity() + kvstart) / NMETA;
//对指定范围的数据进行排序,默认使用的QuickSort
sorter.sort(MapOutputBuffer.this, mstart, mend, reporter);

      对数据排完序后就需要将已排序数据写出到文件中,源代码如下:

int spindex = mstart;
//记录索引的startOffset、rawLength和partLength
final IndexRecord rec = new IndexRecord();
//封装value的字节表示的内部类
final InMemValBytes value = http://www.mamicode.com/new InMemValBytes();>

      当保存索引的缓存超过限制时,就将索引保存到文件中,源代码如下:

if (totalIndexCacheMemory >= indexCacheMemoryLimit) {
 // create spill index file
//MAP_OUTPUT_INDEX_RECORD_LENGTH值为24,表示索引文件中每条记录的大小
 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;
}

      综合上面的分析可知,当在map方法中执行context.write时,将先数据写入到缓存中,当缓存中的数据达到预先设置的阈值时由后台SpillThread线程负责数据排序并将数据溢出到map任务的中间输出文件中。

Hadoop-2.4.1学习之Map任务源码分析(下)