我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。
在文章《 MapReduce V1:TaskTracker设计要点概要分析 》中我们已经了解了org.apache.hadoop.mapred.Child启动的基本流程,在Child VM启动的过程中会运行MapTask,实际是运行用户编写的MapReduce程序中的map方法中的处理逻辑,我们首先看一下,在Child类中,Child基于TaskUmbilicalProtocol协议与TaskTracker通信,获取到该Child VM需要加载的Task相关数据,包括Task本身,代码如下所示:
final TaskUmbilicalProtocol umbilical = taskOwner.doAs(new PrivilegedExceptionAction<TaskUmbilicalProtocol>() { @Override public TaskUmbilicalProtocol run() throws Exception { // 建立Child到TaskTracker的RPC连接 return (TaskUmbilicalProtocol)RPC.getProxy(TaskUmbilicalProtocol.class, TaskUmbilicalProtocol.versionID, address, defaultConf); } }); ... ... JvmContext context = new JvmContext(jvmId, pid); // 根据启动Child VM命令行传递的参数构造一个JvmContext对象 ... ... JvmTask myTask = umbilical.getTask(context); // 基于umbilical获取到一个JvmTask ... ... task = myTask.getTask(); // 通过JvmTask获取到MapTask或ReduceTask
上面代码中,JvmTask中就包含了一个Task,也就是task,它可能是MapTask或ReduceTask。看一下在org.apache.hadoop.mapred.Child中运行Task的基本代码,如下所示:
// Create a final reference to the task for the doAs block final Task taskFinal = task; childUGI.doAs(new PrivilegedExceptionAction<Object>() { @Override public Object run() throws Exception { try { // use job-specified working directory FileSystem.get(job).setWorkingDirectory(job.getWorkingDirectory()); taskFinal.run(job, umbilical); // 这行是核心代码,运行实际的MapTask或ReduceTask } finally { TaskLog.syncLogs (logLocation, taskid, isCleanup, logIsSegmented(job)); TaskLogsTruncater trunc = new TaskLogsTruncater(defaultConf); trunc.truncateLogs(new JVMInfo( TaskLog.getAttemptDir(taskFinal.getTaskID(), taskFinal.isTaskCleanupTask()), Arrays.asList(taskFinal))); } return null; } });
我们关注执行MapTask,上面,通过调用MapTask的run方法,来实际启动MapTask的运行。
MapTask运行的整体流程,如下图所示:
上面流程比较直观,我们结合MapTask的run方法的代码,来进行分析,代码如下所示:
@Override public void run(final JobConf job, final TaskUmbilicalProtocol umbilical) throws IOException, ClassNotFoundException, InterruptedException { this.umbilical = umbilical; // start thread that will handle communication with parent TaskReporter reporter = new TaskReporter(getProgress(), umbilical, jvmContext); // 创建TaskReporter对象 reporter.startCommunicationThread(); boolean useNewApi = job.getUseNewMapper(); initialize(job, getJobID(), reporter, useNewApi); // 初始化MapTask // check if it is a cleanupJobTask if (jobCleanup) { runJobCleanupTask(umbilical, reporter); // 运行JobCleanupTask return; } if (jobSetup) { runJobSetupTask(umbilical, reporter); // 运行JobSetupTask return; } if (taskCleanup) { runTaskCleanupTask(umbilical, reporter); // 运行TaskCleanupTask return; } if (useNewApi) { runNewMapper(job, splitMetaInfo, umbilical, reporter); // 运行MapTask } else { runOldMapper(job, splitMetaInfo, umbilical, reporter); } done(umbilical, reporter); // Task运行完成 }
上面代码中,run方法的参数TaskUmbilicalProtocol umbilical表示一个RPC代理对象,通过该对象可以与TaskTracker进行通信,从而在Task运行过程中,能够将Task的运行进度、状态信息汇报给TaskTracker。
通过上面看出,启动一个MapTask,可能运行的是JobCleanupTask、JobSetupTask、TaskCleanupTask、Mapper四种Task之中的一种,运行每种Task都会向TaskTracker进行汇报。关于一个MapTask如何被划分的,可以参考JobTracker端在JobInProgress中initTasks()方法。
下面,我们根据MapTask中run方法中的处理流程,分为如下几个子流程,进行详细分析:
这里,主要分析在MapTask的run方法中,调用initialize方法的初始化逻辑。先看在调用initialize方法之前,首先创建了一个TaskReporter线程对象,该对象又是基于TaskUmbilicalProtocol umbilical来实现将Task运行状态汇报给TaskTracker。在initialize方法中的初始化流程如下所示:
执行JobCleanupTask,主要是清理Job运行过程中产生的数据,因为该Task可能上次运行过一次,但是失败,为了下一次重新运行,需要将之前失败Task的数据清理掉。即使Job运行成功,也需要清理MapTask执行后输出的中间结果数据,具体流程,如下图所示:
上面的序列图比较详细,将Task运行过程中与TaskTracker之间进行通信都描述出来,我们总结如下几个要点:
运行JobSetupTask,主要是初始化Job对应的临时目录,为后续运行Task做准备,具体处理流程,如下图所示:
在创建运行Job的基本临时目录以后,也需要与TaskTracker通信,汇报该Task的运行状态。
TaskCleanupTask与JobCleanupTask类似,主要是清理Task运行过程中产生的一些临时目录和文件,具体流程如下图所示:
MapTask执行的核心逻辑在runNewMapper方法中,该方法中对应的处理流程,如下图所示:
方法runNewMapper的声明,如下所示:
private <INKEY,INVALUE,OUTKEY,OUTVALUE> void runNewMapper(final JobConf job, final TaskSplitIndex splitIndex, final TaskUmbilicalProtocol umbilical, TaskReporter reporter ) throws IOException, ClassNotFoundException, InterruptedException
其中,JobConf job包含了该Job的配置信息,TaskSplitIndex splitIndex包含了该MapTask所处理的InputSplit的信息(包括splitLocation和startOffset),TaskUmbilicalProtocol umbilical是与TaskTracker通信的代理对象,TaskReporter reporter是一个与TaskTracker通信的线程。上面的序列图,所描述的具体处理流程,如下所示:
下面,我们详细看一下,Mapper处理过程中相关的要点:
Mapper.Context是Mapper类的一个内部类,它包含了运行一个MapTask过程中所需要的所有上下文信息,该类的继承层次结构,如下图所示:
可以看出,一个Mapper对应的执行上下文信息,继承了该Mapper对应TaskAttempt的上下文信息,再向上继承了Job的上下文信息,一个Job包含的配置信息都可以被一个Mapper读取到。
MapperContext中包含如下信息:
private RecordReader<KEYIN,VALUEIN> reader; private InputSplit split;
通过使用一个RecordReader,能够读取InputSplit对应的HDFS上的Block文件数据。TaskInputOutputContext中包含如下信息:
private RecordWriter<KEYOUT,VALUEOUT> output; private StatusReporter reporter; private OutputCommitter committer;
可见,在该层能够实现将Task运行的统计,通过StatusReporter以Counter的形式收集,并提供进度(Progress)获取接口。同时,使用RecordWriter将Mapper的输出写入到文件系统。OutputCommitter是一个比较重要的对象,该抽象类代码,如下所示:
public abstract class OutputCommitter { public abstract void setupJob(JobContext jobContext) throws IOException; public void commitJob(JobContext jobContext) throws IOException { cleanupJob(jobContext); } @Deprecated public void cleanupJob(JobContext context) throws IOException { } public void abortJob(JobContext jobContext, JobStatus.State state) throws IOException { cleanupJob(jobContext); } public abstract void setupTask(TaskAttemptContext taskContext) throws IOException; public abstract boolean needsTaskCommit(TaskAttemptContext taskContext) throws IOException; public abstract void commitTask(TaskAttemptContext taskContext) throws IOException; public abstract void abortTask(TaskAttemptContext taskContext) throws IOException; }
可以参考实现类FileOutputCommitter,它主要负责,在Job运行过程中,管理Task执行过程中对应的文件或目录的信息,如开始运行之前创建目录,运行完成后将临时有用的文件移动到供Job共享的目录下,也会执行无用临时文件的清理。
TaskAttemptContext中主要是有关一个TaskAttempt的相关信息,包括TaskAttemptID及其状态信息。
JobContext主要包括一个Job的相关配置信息,如MapReduce Job的输入输出规格(输入输出KV格式)、Mapper类信息、输入文件格式信息,还有JobID信息。
通过调用mapper.run(mapperContext);进入Mapper的执行流程,我们首先看一下,Mapper提供的接口的声明,代码如下所示:
public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> { public class Context extends MapContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT> { // Mapper.Context是Mapper运行时上下文对象,包含了运行MapTask所需要的基本信息 public Context(Configuration conf, TaskAttemptID taskid, RecordReader<KEYIN,VALUEIN> reader, RecordWriter<KEYOUT,VALUEOUT> writer, OutputCommitter committer, StatusReporter reporter, InputSplit split) throws IOException, InterruptedException { super(conf, taskid, reader, writer, committer, reporter, split); } } /** * Called once at the beginning of the task. */ protected void setup(Context context) throws IOException, InterruptedException { // 执行map方法之前,进行初始化工作 // NOTHING } /** * Called once for each key/value pair in the input split. Most applications * should override this, but the default is the identity function. */ @SuppressWarnings("unchecked") protected void map(KEYIN key, VALUEIN value, Context context) throws IOException, InterruptedException { // 用户实际实现的map处理逻辑 context.write((KEYOUT) key, (VALUEOUT) value); } /** * Called once at the end of the task. */ protected void cleanup(Context context ) throws IOException, InterruptedException { // 执行map完成之后,进行清理 // NOTHING } /** * Expert users can override this method for more complete control over the * execution of the Mapper. * @param context * @throws IOException */ public void run(Context context) throws IOException, InterruptedException { // 该方法是程序的驱动入库,循环调用map方法处理每一个键值对数据 setup(context); try { while (context.nextKeyValue()) { map(context.getCurrentKey(), context.getCurrentValue(), context); } } finally { cleanup(context); } } }
上面run方法中,输入的InputSplit对应的每一个记录,循环调用map方法,map方法中是用户实现的MapReduce的Mapper中的代码。
在runNewMapper方法中可以看到,如果Job有0个ReduceTask,则Mapper将结果直接写入到HDFS文件;而如果Job有大于0个ReduceTask,则Mapper将结果输出到本地,是中间结果。代码如下所示:
if (job.getNumReduceTasks() == 0) { output = new NewDirectOutputCollector(taskContext, job, umbilical, reporter); } else { output = new NewOutputCollector(taskContext, job, umbilical, reporter); }
上面,NewDirectOutputCollector实际上将Mapper的输出直接写入到HDFS文件;而NewOutputCollector将Mapper输出写入到本地文件,我们看一下NewOutputCollector是如何实现Mapper输出落地的,它的构造方法,代码如下所示:
NewOutputCollector(org.apache.hadoop.mapreduce.JobContext jobContext, JobConf job, TaskUmbilicalProtocol umbilical, TaskReporter reporter) throws IOException, ClassNotFoundException { collector = new MapOutputBuffer<K,V>(umbilical, job, reporter); // partitions = jobContext.getNumReduceTasks(); // 根据ReduceTask个数计算Mapper输出分区个数 if (partitions > 0) { // 分区个数大于0,创建一个Partitioner对象 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 -1; } }; } }
可以看出,上面的collector里面的逻辑非常核心,它是MapOutputBuffer类实例,该类实现也是相对复杂的。接着,我们从如下几个方面详细分析:
Mapper输出后,会首先将键值对数据写入到一个Buffer中,通过其它的几个Buffer来跟踪键值对的在Buffer中偏移位置信息,以及键值对所属的分区(Partition)。我们看一下NewOutputCollector调用write方法输出键值对数据的逻辑:
@Override public void write(K key, V value) throws IOException, InterruptedException { collector.collect(key, value, partitioner.getPartition(key, value, partitions)); }
通过partitioner计算一个输出键值对所属的分区,然后调用MapOutputBuffer的collect方法实现Mapper中间结果数据输出。下面,我们看MapOutputBuffer涉及到的一些存储结构,其中包含3个buffer,如下代码所示:
private final int[] kvoffsets; // indices into kvindices private final int[] kvindices; // partition, k/v offsets into kvbuffer private byte[] kvbuffer;
上面的数组结构,kvbuffer用来存储Map输出的键值对数据,kvoffsets用来存储Map输出的键值对数据在kvbuffer中的偏移位置,kvindices用来记录一个键值对对应的分区、键开始、值开始位置的数据结构,具体我们会在后面详细说明。首先,看一下用来定义和限制这些buffer的一些基本默认配置参数:
io.sort.mb = 100MB io.sort.record.percent = 0.05 io.sort.spill.percent = 0.8 sortmb = 100MB rEcper = 0.05 spillper = 0.8 RECSIZE = 16 ACCTSIZE = 3
基于这些配置的参数,计算得到一些buffer的大小及其限制配置,为了直观我们通过使用MB来表示大小,如下所示:
maxMemUsage = sortmb <<20 = 104857600 = 100MB recordCapacity = maxMemUsage * recper = 104857600 * 0.05 = 5242880 = 5MB recordCapacity -= recordCapacity % RECSIZE = 5242880 - 5242880 % 16 = 5242880 = 5MB maxMemUsage – recordCapacity = 104857600 - 5242880 = 99614720 = 95MB softBufferLimit = ((maxMemUsage – recordCapacity) * spillper) = 99614720 * 0.8 = 79691776 = 76MB softRecordLimit = (recordCapacity * spillper) = 5242880 * 0.8 = 4194304 = 4MB
现在,我们看一下这些buffer的结构即相关配置参数,如下图所示:
Mapper分别读取输入Split的每一行数据,然后输出键值对数据,这些键值对数据在上面讲述到的Buffer中的存储细节,如下图所示:
上图中,输出了2个键值对数据,首先kvbuffer中会顺序存储每一个键值对数据,同时,kvoffsets会记录每一个键值对的顺序编号,或者说,在kvbuffer中的偏移位置,在kvindices中包含了每一个键值对的三个属性信息:所属分区(kvi partition)、键起始偏移量(ki start)、值起始偏移量(vi start)。
当内存中的键值对数据的存储量在上述Buffer中达到一定限制(通过配置io.sort.spill.percent决定,默认是大约95MB的Buffer存储容量达到80%,即大约76MB时会触发将Buffer中数据写入磁盘文件)时,就会通过SpillThread线程,将Buffer中的数据写入本地磁盘文件中,可以在SpillThread类中的sortAndSpill方法中看到具体实现。在将Buffer中的键值对数据写入磁盘之前,先进行一次内存排序,排序的规则是:MapOutputBuffer内部有3个Buffer,排序是对键值对偏移位置的Buffer kvoffsets进行排序,保证每一个键值对所属的分区(Partition)按照升序排序,然后再保证每个分区(Partition)中的键值对按照键进行排序,这样最后得到的kvoffsets中的键值对数据就是按照分区进行分组,并且每个分组中是按照键排过序的。
如果用户编写的MapReduce程序,指定了Combiner,则再排序之后,写入磁盘文件之前,调用Combiner对数据进行合并,具体可以参考CombinerRunner实现类。
经过几次填充Buffer,达到配置写入容量值会写入文件,最终得到多个分区的、按照键(Key)排序的文件,在所有数据都输出结束后,会对上述生成的多个文件进行合并,合并成一个大文件,该大文件也是经过分区的、排序的文件。在最后的合并阶段,也会检查用户是否配置了Combiner,如果是则会对每个分区的数据做一个局部化简合并,输出到最终的一个文件中,并将多个小的分区文件删除。
本文基于 署名-非商业性使用-相同方式共享 4.0 许可协议发布,欢迎转载、使用、重新发布,但务必保留文章署名时延军(包含链接:http://shiyanjun.cn),不得用于商业目的,基于本文修改后的作品务必以相同的许可发布。如有任何疑问,请与我联系。