MapReduce框架源码解析

基础环境

Version: Hadoop v3.1.2 Example: MultiFileWordCount.java

先看一下整体的流程图，之后分析源码，加深对这张图的理解以及了解图中一些细节的实现。 由于整个流程涉及的代码很多，这篇只会分析1-5a提交任务到YARN部分，5b-11会在之后的YARN源码解析的文章中详细介绍。

源码解析

程序入口，可以看到实例化了MultiFileWordCount对象并调用了ToolRunner的run方法，非常简单。 MultiFileWordCount，继承了Configured，为了使用getConf和setConf等Configuration相关的操作。实现了Tool接口，重写了Tool.run方法。

public class MultiFileWordCount extends Configured implements Tool {

ToolRunner.run

然后就是调用MultiFileWordCount中重写的run方法：

MyInputFormat.class

在MyInputFormat中实现了createRecordReader方法，实例化了一个CombineFileRecordReader对象，该类可以根据大小合并多个文件块到一个map读。

Text.class

设置了输出的key的类，Text遵循UTF-8，支持序列化和反序列化，并且提供一些compare、getLength等函数。 这个代码就不贴了，是Hadoop基础包中的一些函数，有兴趣可以去看下org/apache/hadoop/io/Text.java

IntWritable.class

设置了输出的value的类，封装了int，包含了compare、write等方法。 同属于Hadoop基础包中的一些函数，可以参看org/apache/hadoop/io/IntWritable.java

MapClass.class

map阶段实现很简单，继承了框架接口Mapper并重写了map方法。

IntSumReducer.class

reduce阶段也非常简单，继承了框架接口Reducer并重写了reduce方法。

job.waitForCompletion

提交MapReduce任务入口。

Job.submit

生成JobSubmitter对象并将Job提交到YARN

JobSubmitter.submitJobInternal

提交Job，核心方法之一，大致分为以下几个步骤： 1、校验该job的input和output相关信息的正确性。 2、计算舒服输入的分片/map数量（input splits）。 3、设置更新需要公用的job conf 4、把job conf和相关jar包copy到HDFS之前建的MapReduce目录下。 5、提交job，监听job状态。

YARNRunner.submitJob

核心方法是resMgrDelegate.submitApplication(appContext)

YarnClientImpl.submitApplication

总结

以上就是MapReduce提交任务到YARN以及监听任务状态的主要代码了，断断续续写了一周，深刻体会到了Hadoop代码对一些细节的控制真的非常全面、精简，让人看似觉得少考虑了一些情况，但细细一想缺找不出漏洞。 在后续的文章中，会对Hive以及YARN的源码进行分析Hadoop源码系列文章目录。