Hadoop源代码分析（MapTask辅助类 I）

 

原博客地址http://caibinbupt.iteye.com/blog/401374

 http://www.cnblogs.com/ggjucheng/archive/2013/02/20/2917799.html

mapTask的辅助类主要针对Mapper的输入和输出。首先我们来看MapTask中用的的Mapper输入，在类图中，这部分位于右上角。

MapTask.TrackedRecordReader是一个Wrapper，在原有输入RecordReader的基础上，添加了收集上报统计数据的功能。

MapTask.SkippingRecordReader也是一个Wrapper，它在 MapTask.TrackedRecordReader的基础上，添加了忽略部分输入的功能。在分析 MapTask.SkippingRecordReader之前，我们先看一下类SortedRanges和它相关的类。

 

 

类SortedRanges.Ranges 表示了一个范围，以开始位置和范围长度（这样的话就可以表示长度为0的范围）来表示一个范围，并提供了一系列的范围操作方法。注意，方法 getEndIndex得到的右端点并不包含在范围内（应理解为开区间）。SortedRanges包含了一系列不重叠的范围，为了保证包含的范围不重 叠，在add方法和remove方法上需要做一些处理，保证不重叠的约束。SkipRangeIterator是访问SortedRanges包含的 Ranges的迭代器。

MapTask.SkippingRecordReader的实现很简单，因为要忽略的输入都保持在 SortedRanges.Ranges，只需要在next方法中，判断目前范围时候落在SortedRanges.Ranges中，如果是，忽略，并将 忽略的记录写文件（可配置）

NewTrackingRecordReader和NewOutputCollector被新API使用，我们不分析。

MapTask的输出辅助类都继承自MapOutputCollector，它只是在OutputCollector的基础上添加了close和flush方法。

DirectMapOutputCollector用在Reducer的数目为0，就是不需要Reduce阶段的时候。它是直接通过

out = job.getOutputFormat().getRecordWriter(fs, job, finalName, reporter);

得到对应的RecordWriter，collect直接到RecordWriter上。

如果Mapper后续有reduce任务，系统会使用MapOutputBuffer做为输出，这是个比较复杂的类，有1k行左右的代码。

我们知道，Mapper是通过 OutputCollector将Map的结果输出，输出的量很大，Hadoop的机制是通过一个circle buffer 收集Mapper的输出, 到了io.sort.mb * percent量的时候，就spill到disk，如下图。图中出现了两个数组和一个缓冲区，kvindices保持了记录所属的（Reduce）分 区，key在缓冲区开始的位置和value在缓冲区开始的位置，通过kvindices，我们可以在缓冲区中找到对应的记录。kvoffets用于在缓冲 区满的时候对kvindices的partition进行排序，排完序的结果将输出到输出到本地磁盘上，其中索引（kvindices）保持在 spill{spill号}.out.index中，数据保存在spill{spill号}.out中。

 

 

 

当Mapper任务结束后，有可能会出现多个spill文件，这些文件会做一个归并排序，形成Mapper的一个输出（spill.out和spill.out.index），如下图：

 

 

这个输出是按partition排序的，这样的话，Mapper的输出被分段，Reducer要获取的就是spill.out中的一段。（注意，内存和硬盘上的索引结构不一样）

 

（感谢彭帅的Hadoop Map Stage流程分析 http://www.cnblogs.com/OnlyXP/archive/2009/05/25/1488811.html）