mapreduce 编程模型

MapReduce是在总结大量应用的共同特点的基础上抽象出来的分布式计算框架，它适用的应用场景往往具有一个共同的特点：任务可被分解成相互独立的子问题。基于该特点，MapReduce编程模型给出了其分布式编程方法，共分5个步骤：

1）迭代（iteration）。遍历输入数据，并将之解析成key/value对。

2）将输入key/value对映射（map）成另外一些key/value对。

3）依据key对中间数据进行分组（grouping）。

4）以组为单位对数据进行归约（reduce）。

5）迭代。将最终产生的key/value对保存到输出文件中。

MapReduce将计算过程分解成以上5个步骤带来的最大好处是组件化与并行化。

为了实现MapReduce编程模型，Hadoop设计了一系列对外编程接口。用户可通过实现这些接口完成应用程序的开发。

MapReduce编程接口体系结构

MapReduce编程模型对外提供的编程接口体系结构如图所示，整个编程模型位于应用程序层和MapReduce执行器之间，可以分为两层。第一层是最基本的Java API，主要有5个可编程组件，分别是InputFormat、Mapper、Partitioner、Reducer和OutputFormat 。Hadoop自带了很多直接可用的InputFormat、Partitioner和OutputFormat，大部分情况下，用户只需编写Mapper和Reducer即可。第二层是工具层，位于基本Java API之上，主要是为了方便用户编写复杂的MapReduce程序和利用其他编程语言增加MapReduce计算平台的兼容性而提出来的。在该层中，主要提供了4个编程工具包:

JobControl：方便用户编写有依赖关系的作业，这些作业往往构成一个有向图，所以通常称为DAG（Directed Acyclic Graph）作业。

ChainMapper/ChainReducer：方便用户编写链式作业，即在Map或者Reduce阶段存在多个Mapper，形式如：

[MAPPER+ REDUCER MAPPER*]

Hadoop Streaming：方便用户采用非Java语言编写作业，允许用户指定可执行文件或者脚本作为Mapper/Reducer。

Hadoop Pipes：专门为C/C++程序员编写MapReduce程序提供的工具包

作业配置与提交

1. Hadoop 配置文件介绍

在Hadoop中，Common、HDFS和MapReduce各有对应的配置文件，用于保存对应模块中可配置的参数。这些配置文件均为XML格式且由两部分构成：系统默认配置文件和管理员自定义配置文件。其中，系统默认配置文件分别是core-default.xml、hdfs-default.xml和mapred-default.xml，它们包含了所有可配置属性的默认值。而管理员自定义配置文件分别是core-site.xml、hdfs-site.xml和mapred-site.xml。它们由管理员设置，主要用于定义一些新的配置属性或者覆盖系统默认配置文件中的默认值。通常这些配置一旦确定，便不能被修改（如果想修改，需重新启动Hadoop）。需要注意的是，core-default.xml和core-site.xml属于公共基础库的配置文件，默认情况下，Hadoop总会优先加载它们。在Hadoop中，每个配置属性主要包括三个配置参数：name、value和description，分别表示属性名、属性值和属性描述。其中，属性描述仅仅用来帮助用户理解属性的含义，Hadoop内部并不会使用它的值。此外，Hadoop为配置文件添加了两个新的特性：final参数和变量扩展。

final参数：如果管理员不想让用户程序修改某些属性的属性值，可将该属性的final参数置为true，比如：

<property>
<name>mapred.map.tasks.speculative.execution</name>
<value>true</value>
<final>true</final>
</property>

管理员一般在XXX-site.xml配置文件中为某些属性添加final参数，以防止用户在应用程序中修改这些属性的属性值。

2. MapReduce作业配置与提交

在MapReduce中，每个作业由两部分组成：应用程序和作业配置。其中，作业配置内容包括环境配置和用户自定义配置两部分。环境配置由Hadoop自动添加，主要由mapreddefault.xml和mapred-site.xml两个文件中的配置选项组合而成；用户自定义配置则由用户自己根据作业特点个性化定制而成，比如用户可设置作业名称，以及Mapper/Reducer、Reduce Task个数等。在新旧两套API中，作业配置接口发生了变化，首先通过一个例子感受一下使用上的不同。

旧API作业配置实例：

JobConf job = new JobConf(new Configuration(), MyJob.class);
job.setJobName("myjob");
job.setMapperClass(MyJob.MyMapper.class);
job.setReducerClass(MyJob.MyReducer.class);
JobClient.runJob(job);
新API作业配置实例：
Configuration conf = new Configuration();
Job job = new Job(conf, "myjob ");
job.setJarByClass(MyJob.class);
job.setMapperClass(MyJob.MyMapper.class);
job.setReducerClass(MyJob.MyReducer.class);
System.exit(job.waitForCompletion(true) ? 0 : 1);

从以上两个实例可以看出，新版API用Job类代替了JobConf和JobClient两个类，这样，仅使用一个类的同时可完成作业配置和作业提交相关功能，进一步简化了作业编写方式。

InputFormat接口的设计与实现

InputFormat主要用于描述输入数据的格式，它提供以下两个功能。

数据切分：按照某个策略将输入数据切分成若干个split，以便确定Map Task个数以及对应的split。

为Mapper提供输入数据：给定某个split，能将其解析成一个个key/value对。

旧版API的InputFormat解析如图所示，在旧版API中，InputFormat是一个接口，它包含两种方法：

InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;
RecordReader<K, V> getRecordReader(InputSplit split,JobConf job, Reporter reporter) throws IOException;

getSplits方法主要完成数据切分的功能，它会尝试着将输入数据切分成numSplits个

InputSplit。InputSplit有以下两个特点。

逻辑分片：它只是在逻辑上对输入数据进行分片，并不会在磁盘上将其切分成分片进行存储。InputSplit只记录了分片的元数据信息，比如起始位置、长度以及所在的节点列表等。

可序列化：在Hadoop中，对象序列化主要有两个作用：进程间通信和永久存储。此处，InputSplit支持序列化操作主要是为了进程间通信。作业被提交到JobTracker之前，Client会调用作业InputFormat中的getSplits函数，并将得到的InputSplit序列化到文件中。这样，当作业提交到JobTracker端对作业初始化时，可直接读取该文件，解析出所有InputSplit，并创建对应的Map Task

getRecordReader方法返回一个RecordReader对象，该对象可将输入的InputSplit

解析成若干个key/value对。MapReduce框架在Map Task执行过程中，会不断调用RecordReader对象中的方法，迭代获取key/value对并交给map()函数处理，主要代码（经过简化）如下：

//调用InputSplit的getRecordReader方法获取RecordReader<K1, V1> input
……
K1 key = input.createKey();
V1 value = input.createValue();
while (input.next(key, value)) {
//调用用户编写的map()函数
}
input.close();

前面分析了InputFormat接口的定义，接下来介绍系统自带的各种InputFormat实现。

为了方便用户编写MapReduce程序，Hadoop自带了一些针对数据库和文件的InputFormat实现，具体如图所示。通常而言，用户需要处理的数据均以文件形式存储到HDFS上，所以我们重点针对文件的InputFormat实现进行讨论

如图所示，所有基于文件的InputFormat实现的基类是FileInputFormat，并由此派生出针对文本文件格式的TextInputFormat、KeyValueTextInputFormat和NLineInputFormat，针对二进制文件格式的SequenceFileInputFormat等。整个基于文件的InputFormat体系的设计思路是，由公共基类FileInputFormat采用统一的方法对各种输入文件进行切分，比如按照某个固定大小等分，而由各个派生InputFormat自己提供机制将进一步解析InputSplit。对应到具体的实现是，基类FileInputFormat提供getSplits实现，而派生类提供getRecordReader实现。

为了帮助读者深入理解这些InputFormat的实现原理，我们选取TextInputFormat与SequenceFileInputFormat进行重点介绍。

我们首先介绍基类FileInputFormat的实现。它最重要的功能是为各种InputFormat提供统一的getSplits函数。该函数实现中最核心的两个算法是文件切分算法和host选择算法.

（1）文件切分算法

文件切分算法主要用于确定InputSplit的个数以及每个InputSplit对应的数据段。

FileInputFormat以文件为单位切分生成InputSplit。对于每个文件，由以下三个属性值确定其对应的InputSplit的个数。

goalSize：它是根据用户期望的InputSplit数目计算出来的，即totalSize/numSplits。其中，totalSize为文件总大小；numSplits为用户设定的Map Task个数，默认情况下是1。

minSize：InputSplit的最小值，由配置参数mapred.min.split.size确定，默认是1。

blockSize：文件在HDFS中存储的block大小，不同文件可能不同，默认是64 MB。

这三个参数共同决定InputSplit的最终大小，计算方法如下：

splitSize = max{minSize, min{goalSize, blockSize}}

一旦确定splitSize值后，FileInputFormat将文件依次切成大小为splitSize的InputSplit，最后剩下不足splitSize的数据块单独成为一个InputSplit。

输入目录下有三个文件file1、file2和file3，大小依次为1 MB，32 MB和250 MB。若blockSize采用默认值64 MB，则不同minSize和goalSize下，file3切分结果如表所示（三种情况下，file1与file2切分结果相同，均为1个InputSplit）。

如果想让InputSplit尺寸大于block尺寸，则直接增大配置参数mapred.min.split.size即可。

（2）host选择算法

待InputSplit切分方案确定后，下一步要确定每个InputSplit的元数据信息。这通常由四部分组成：<file, start, length, hosts>，分别表示InputSplit所在的文件、起始位置、长度以及所在的host（节点）列表。其中，前三项很容易确定，难点在于host列表的选择方法。

InputSplit的host列表选择策略直接影响到运行过程中的任务本地性。介绍Hadoop架构时，我们提到HDFS上的文件是以block为单位组织的，一个大文件对应的block可能遍布整个Hadoop集群，而InputSplit的划分算法可能导致一个InputSplit对应多个block ，这些block可能位于不同节点上，这使得Hadoop不可能实现完全的数据本地性。为此，Hadoop将数据本地性按照代价划分成三个等级：node locality、rack locality和datacenter locality（Hadoop还未实现该locality级别）。在进行任务调度时，会依次考虑这3个节点的locality，即优先让空闲资源处理本节点上的数据，如果节点上没有可处理的数据, 则处理同一个机架上的数据，最差情况是处理其他机架上的数据（但是必须位于同一个数据中心）。

虽然InputSplit对应的block可能位于多个节点上，但考虑到任务调度的效率，通常不会把所有节点加到InputSplit的host列表中，而是选择包含（该InputSplit）数据总量最大的前几个节点（Hadoop限制最多选择10个，多余的会过滤掉），以作为任务调度时判断任务是否具有本地性的主要凭证。为此，FileInputFormat设计了一个简单有效的启发式算法：

首先按照rack包含的数据量对rack进行排序，然后在rack内部按照每个node包含的数据量对node排序，最后取前N个node的host作为InputSplit的host列表，这里的N为block副本数。这样，当任务调度器调度Task时，只要将Task调度给位于host列表的节点，就认为该Task满足本地性。

从host选择算法可知，当InputSplit尺寸大于block尺寸时，Map Task并不能实现完全数据本地性，也就是说，总有一部分数据需要从远程节点上读取，因而可以得出以下结论：

当使用基于FileInputFormat实现InputFormat时，为了提高Map Task的数据本地性，应尽量使InputSplit大小与block大小相同。

分析完FileInputFormat实现方法，接下来分析派生类TextInputFormat与SequenceFileInputFormat的实现。

前面提到，由派生类实现getRecordReader函数，该函数返回一个RecordReader对象。它实现了类似于迭代器的功能，将某个InputSplit解析成一个个key/value对。在具体实现时，RecordReader应考虑以下两点:

(1)定位记录边界：为了能够识别一条完整的记录，记录之间应该添加一些同步标识.对于TextInputFormat，每两条记录之间存在换行符；对于SequenceFileInputFormat，每隔若干条记录会添加固定长度的同步字符串。通过换行符或者同步字符串，它们很容易定位到一个完整记录的起始位置。另外，由于FileInputFormat仅仅按照数据量多少对文件进行切分，因而InputSplit的第一条记录和最后一条记录可能会被从中间切开。为了解决这种记录跨越InputSplit的读取问题，RecordReader规定每个InputSplit的第一条不完整记录划给前一个InputSplit处理。

(2)解析key/value：定位到一条新的记录后，需将该记录分解成key和value两部分。对于TextInputFormat，每一行的内容即为value，而该行在整个文件中的偏移量为key。对于SequenceFileInputFormat，每条记录的格式为：

[record length] [key length] [key] [value]

其中，前两个字段分别是整条记录的长度和key的长度，均为4字节，后两个字段分别是key和value的内容。知道每条记录的格式后，很容易解析出key和value。

OutputFormat接口的设计与实现

OutputFormat主要用于描述输出数据的格式，它能够将用户提供的key/value对写入特定格式的文件中。本小节将介绍Hadoop如何设计OutputFormat接口，以及一些常用的OutputFormat实现。

旧版API的OutputFormat解析如图所示，在旧版API中，OutputFormat是一个接口，它包含两个方法：

RecordWriter<K, V> getRecordWriter(FileSystem ignored, JobConf job,
String name, Progressable progress)throws IOException;void checkOutputSpecs(FileSystem ignored, JobConf job) throws IOException;

checkOutputSpecs方法一般在用户作业被提交到JobTracker之前，由JobClient自动调用，以检查输出目录是否合法。

getRecordWriter方法返回一个RecordWriter类对象。该类中的方法write接收一个key/value对，并将之写入文件。在Task执行过程中，MapReduce框架会将map()或者reduce()函数产生的结果传入write方法，主要代码（经过简化）如下。

假设用户编写的map()函数如下：

public void map(Text key, Text value,
OutputCollector<Text, Text> output, 
Reporter reporter) throws IOException {
// 根据当前key/value产生新的输出<newKey, newValue>，并输出
……
output.collect(newKey, newValue);
}

则函数output.collect(newKey, newValue)内部执行代码如下：

RecordWriter<K, V> out = job.getOutputFormat().getRecordWriter(...);
out.write(newKey, newValue);

Hadoop自带了很多OutputFormat实现，它们与InputFormat实现相对应，具体如图所示。所有基于文件的OutputFormat实现的基类为FileOutputFormat，并由此派生出一些基于文本文件格式、二进制文件格式的或者多输出的实现。

为了深入分析OutputFormat的实现方法，我们选取比较有代表性的FileOutputFormat类进行分析。同介绍InputFormat实现的思路一样，我们先介绍基类FileOutputFormat，再介绍其派生类TextOutputFormat。

基类FileOutputFormat需要提供所有基于文件的OutputFormat实现的公共功能，总结起来，主要有以下两个：

（1）实现checkOutputSpecs接口

该接口在作业运行之前被调用，默认功能是检查用户配置的输出目录是否存在，如果存在则抛出异常，以防止之前的数据被覆盖。

（2）处理side-effect file

任务的side-effect file并不是任务的最终输出文件，而是具有特殊用途的任务专属文件。它的典型应用是执行推测式任务。在Hadoop中，因为硬件老化、网络故障等原因，同一个作业的某些任务执行速度可能明显慢于其他任务，这种任务会拖慢整个作业的执行速度。为了对这种“慢任务”进行优化，Hadoop会为之在另外一个节点上启动一个相同的任务，该任务便被称为推测式任务，最先完成任务的计算结果便是这块数据对应的处理结果。为防止这两个任务同时往一个输出文件中写入数据时发生写冲突，FileOutputFormat会为每个Task的数据创建一个side-effect file，并将产生的数据临时写入该文件，待Task完成后，再移动到最终输出目录中。这些文件的相关操作，比如创建、删除、移动等，均由OutputCommitter完成。它是一个接口， Hadoop提供了默认实现FileOutputCommitter，用户也可以根据自己的需求编写OutputCommitter实现，并通过参数{mapred.output.committer.class}指定。OutputCommitter接口定义以及FileOutputCommitter对应的实现如表所示

Mapper与Reducer解析

旧版API的Mapper/Reducer解析

Mapper/Reducer中封装了应用程序的数据处理逻辑。为了简化接口，MapReduce要求所有存储在底层分布式文件系统上的数据均要解释成key/value的形式，并交给Mapper/Reducer中的map/reduce函数处理，产生另外一些key/value。

Mapper与Reducer的类体系非常类似，我们以Mapper为例进行讲解。Mapper的类图如图所示，包括初始化、Map操作和清理三部分。

（1）初始化

Mapper继承了JobConfigurable接口。该接口中的configure方法允许通过JobConf参数对Mapper进行初始化。

（2）Map操作

MapReduce框架会通过InputFormat中RecordReader从InputSplit获取一个个key/value对，并交给下面的map()函数处理：

void map(K1 key, V1 value, OutputCollector<K2, V2> output, Reporter reporter)throws IOException;

该函数的参数除了key和value之外，还包括OutputCollector和Reporter两个类型的参数，分别用于输出结果和修改Counter值。

（3）清理

Mapper通过继承Closeable接口（它又继承了Java IO中的Closeable接口）获得close方法，用户可通过实现该方法对Mapper进行清理。

MapReduce提供了很多Mapper/Reducer实现，但大部分功能比较简单，具体如图所示。它们对应的功能分别是：

ChainMapper/ChainReducer:用于支持链式作业。IdentityMapper/IdentityReducer：对于输入key/value不进行任何处理，直接输出.InvertMapper：交换key/value位置。RegexMapper：正则表达式字符串匹配。TokenMapper：将字符串分割成若干个token（单词），可用作WordCount的Mapper。  LongSumReducer：以key为组，对long类型的value求累加和。

对于一个MapReduce应用程序，不一定非要存在Mapper。MapReduce框架提供了比Mapper更通用的接口：MapRunnable，如图所示。用户可以实现该接口以定制Mapper的调用方式或者自己实现key/value的处理逻辑，比如，Hadoop Pipes自行实现了MapRunnable，直接将数据通过Socket发送给其他进程处理。提供该接口的另外一个好处是允许用户实现多线程Mapper。

Hadoop工作流

很多情况下，用户编写的作业比较复杂，相互之间存在依赖关系，这种依赖关系可以用有向图表示，我们称之为“工作流”。

JobControl 的实现原理

1. JobControl 编程实例

传统的做法是：为每个作业创建相应的JobConf对象，并按照依赖关系依次（串行）提交各个作业，如下所示：

//为4个作业分别创建JobConf对象
JobConf extractJobConf = new JobConf(ExtractJob.class);
JobConf classPriorJobConf = new JobConf(ClassPriorJob.class);
JobConf conditionalProbilityJobConf = new JobConf(ConditionalProbilityJob.class);
JobConf predictJobConf = new JobConf(PredictJob.class);
...//配置各个JobConf
//按照依赖关系依次提交作业
JobClient.runJob(extractJobConf);
JobClient.runJob(classPriorJobConf);
JobClient.runJob(conditionalProbilityJobConf);
JobClient.runJob(predictJobConf);

如果使用JobControl，则用户只需使用addDepending()函数添加作业依赖关系接口，

JobControl会按照依赖关系调度各个作业，具体代码如下：

Configuration extractJobConf = new Configuration();
Configuration classPriorJobConf = new Configuration();
Configuration conditionalProbilityJobConf = new Configuration();
Configuration predictJobConf = new Configuration();
...//设置各个Configuration
//创建Job对象。注意，JobControl要求作业必须封装成Job对象
Job extractJob = new Job(extractJobConf);
Job classPriorJob = new Job(classPriorJobConf);
Job conditionalProbilityJob = new Job(conditionalProbilityJobConf);
Job predictJob = new Job(predictJobConf);
//设置依赖关系，构造一个DAG作业
classPriorJob.addDepending(extractJob); 
conditionalProbilityJob.addDepending(extractJob);
predictJob.addDepending(classPriorJob);
predictJob.addDepending(conditionalProbilityJob);
//创建JobControl对象，由它对作业进行监控和调度
JobControl JC = new JobControl("Native Bayes");
JC.addJob(extractJob);//把4个作业加入JobControl中
JC.addJob(classPriorJob);
JC.addJob(conditionalProbilityJob);
JC.addJob(predictJob);
JC.run(); //提交DAG作业

在实际运行过程中，不依赖于其他任何作业的extractJob会优先得到调度，一旦运行完成，classPriorJob和conditionalProbilityJob两个作业同时被调度，待它们全部运行完成后，predictJob被调度。

对比以上两种方案，可以得到一个简单的结论：使用JobControl编写DAG作业更加简便，且能使多个无依赖关系的作业并行运行。

2. JobControl 设计原理分析

JobControl由两个类组成：Job和JobControl。其中，Job类封装了一个MapReduce作业及其对应的依赖关系，主要负责监控各个依赖作业的运行状态，以此更新自己的状态，其状态转移图如图3-26所示。作业刚开始处于WAITING状态。如果没有依赖作业或者所有依赖作业均已运行完成，则进入READY状态。一旦进入READY状态，则作业可被提交到Hadoop集群上运行，并进入RUNNING状态。在RUNNING状态下，根据作业运行情况，可能进入SUCCESS或者FAILED状态。需要注意的是，如果一个作业的依赖作业失败，则该作业也会失败，于是形成“多米诺骨牌效应”，后续所有作业均会失败。

JobControl封装了一系列MapReduce作业及其对应的依赖关系。它将处于不同状态的作业放入不同的哈希表中，并按照图3-26所示的状态转移作业，直到所有作业运行完成。

在实现的时候，JobControl包含一个线程用于周期性地监控和更新各个作业的运行状态，调度依赖作业运行完成的作业，提交处于READY状态的作业等。同时，它还提供了一些API用于挂起、恢复和暂停该线程。

ChainMapper/ChainReducer的实现原理

ChainMapper/ChainReducer主要为了解决线性链式Mapper而提出的。也就是说，在Map或者Reduce阶段存在多个Mapper，这些Mapper像Linux管道一样，前一个Mapper的输出结果直接重定向到下一个Mapper的输入，形成一个流水线，形式类似于[MAP+REDUCE MAP*]。图展示了一个典型的ChainMapper/ChainReducer的应用场景：在Map阶段，数据依次经过Mapper1和Mapper2处理；在Reduce阶段，数据经过shuffle和sort后；交由对应的Reducer处理，但Reducer处理之后并没有直接写到HDFS上，而是交给另外一个Mapper处理，它产生的结果写到最终的HDFS输出目录中。

需要注意的是，对于任意一个MapReduce作业，Map和Reduce阶段可以有无限个Mapper，但Reducer只能有一个。也就是说，图所示的计算过程不能使用 ChainMapper/ChainReducer完成，而需要分解成两个MapReduce作业。

1. 编程实例

这里给出ChainMapper/ChainReducer的基本使用方法，具体代码如下：

…
conf.setJobName("chain");
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
JobConf mapper1Conf = new JobConf(false); 
JobConf mapper2Conf = new JobConf(false);
JobConf reduce1Conf = new JobConf(false);
JobConf mapper3Conf = new JobConf(false);
…
ChainMapper.addMapper(conf, Mapper1.class, LongWritable.class, Text.class,Text.
class, Text.class, true, mapper1Conf);
ChainMapper.addMapper(conf, Mapper2.class, Text.class, Text.class,
LongWritable.class, Text.class, false, mapper2Conf);
ChainReducer.setReducer(conf, Reducer.class, LongWritable.class, Text.class,Text.
class, Text.class, true, reduce1Conf);
ChainReducer.addMapper(conf, Mapper3.class, Text.class, Text.class,
LongWritable.class, Text.class, false, null);
JobClient.runJob(conf);

用户通过addMapper在Map/Reduce阶段添加多个Mapper。该函数带有8个输入参数，分别是作业的配置、Mapper类、Mapper的输入key类型、输入value类型、输出key类型、输出value类型、key/value是否按值传递和Mapper的配置。其中，第7个参数需要解释一下：Hadoop MapReduce 有一个约定，函数OutputCollector.collect(key, value)执行期间不应改变key和value的值。这主要是因为函数Mapper.map()调用完OutputCollector.

collect(key, value)之后，可能会再次使用key和value值，如果被改变，可能会造成潜在的错误。为了防止OutputCollector直接对key/value修改，ChainMapper允许用户指定key/value传递方式。如果用户确定key/value不会被修改，则可选用按引用传递，否则按值传递。需要注意的是，引用传递可避免对象拷贝，提高处理效率，但需要确保key/value不会被修改。

2. 实现原理分析

ChainMapper/ChainReducer实现的关键技术点是修改Mapper和Reducer的输出

流，将本来要写入文件的输出结果重定向到另外一个Mapper中。在3.3.4节中提到，结果的输出由OutputCollector管理，因而，ChainMapper/ChainReducer需要重新实现一个OutputCollector完成数据重定向功能。

尽管链式作业在Map和Reduce阶段添加了多个Mapper，但仍然只是一个MapReduce作业，因而只能有一个与之对应的JobConf对象。然而，当用户调用addMapper添加Mapper时，可能会为新添加的每个Mapper指定一个特有的JobConf，为此，ChainMapper/ChainReducer将这些JobConf对象序列化后，统一保存到作业的JobConf中。

当链式作业开始执行的时候，首先将各个Mapper的JobConf对象反序列化，并构造对应的Mapper和Reducer对象，添加到数据结构mappers（List<Mapper>类型）和reducer（Reducer类型）中。ChainMapper中实现的map()函数如下，它调用了第一个Mapper，是后续Mapper的“导火索”。

public void map(Object key, Objectvalue, OutputCollector output,
Reporter reporter) throws IOException {
Mapper mapper = chain.getFirstMap();
if (mapper != null) {
mapper.map(key, value, chain.getMapperCollector(0, output, reporter),reporter);
}
}

chain.getMapperCollector返回一个OutputCollector实现—ChainOutputCollector，它

的collect方法如下：

public void collect(K key, V value) throws IOException {
if (nextMapperIndex < mappers.size()) {
//调用下一个Mapper
nextMapper.map(key, value,
new ChainOutputCollector(nextMapperIndex,
nextKeySerialization,
nextValueSerialization,
output, reporter),
reporter);
} else {
// 如果是最后一个Mapper，则直接调用真正的OutputCollector
output.collect(key, value);
}

Hadoop工作流引擎

前面介绍的JobControl和ChainMapper/ChainReducer仅可看作运行工作流的工具。

它们只具备最简单的工作流引擎功能，比如工作流描述、简单的作业调度等。为了增强Hadoop支持工作流的能力，在Hadoop之上出现了很多开源的工作流引擎，主要可概括为两类：隐式工作流引擎和显式工作流引擎。

隐式工作流引擎在MapReduce之上添加了一个语言抽象层，允许用户使用更简单的方式编写应用程序，比如SQL、脚本语言等。这样，用户无须关注MapReduce的任何细节，降低了用户的学习成本，并可大大提高开发效率。典型的代表有Hive 、Pig 和Cascading 。它们的架构从上往下分为以下三层。

功能描述层：直接面向用户提供了一种简单的应用程序编写方法，比如，Hive使用SQL，Pig使用Pig Latin脚本语言，Cascading提供了丰富的Java API。

作业生成器：作业生成器主要将上层的应用程序转化成一批MapReduce作业。这一批MapReduce存在相互依赖关系，实际上是一个DAG。

调度引擎：调度引擎直接构建于MapReduce环境之上，将作业生成器生成的DAG按照依赖关系提交到MapReduce上运行。

显式工作流引擎直接面向MapReduce应用程序开发者，提供了一种作业依赖关系描述方式，并能够按照这种描述方式进行作业调度。典型的代表有Oozie和Azkaban 。它们的架构如图3-30所示，从上往下分为以下两层。

工作流描述语言：工作流描述语言用于描述作业的依赖关系。Oozie采用了XML，而Azkaban采用了key/value格式的文本文件。需要注意的是，这里的作业不仅仅是指MapReduce作业，还包括Shell命令、Pig脚本等。也就是说，一个MapReduce可能依赖一个Pig脚本或者Shell命令。

调度引擎：同隐式工作流引擎的调度引擎功能相同，即根据作业的依赖关系完成作业调度。