1. OutputCommitters
MapReduce使用一个提交协议来确保作业（job）和任务（task）都完全成功或失败。这个通过 OutputCommiter来实现。
新版本 MapReduce API中，OutputCommitter 由OutputFormat 通过getOutputCommitter() 方法确定。默认为FileOutputCommitter，适用于有文件输出的MapReduce任务。若是需要，也可以实现一个新的OutputCommitter类，以对作业的完成或任务做自定义设置或清理。
OutputCommiter 部分源码如下：
public abstract class OutputCommitter extends org.apache.hadoop.mapreduce.OutputCommitter {
public OutputCommitter() {
}
public abstract void setupJob(JobContext var1) throws IOException;
/** @deprecated */
@Deprecated
public void cleanupJob(JobContext jobContext) throws IOException {
}
public void commitJob(JobContext jobContext) throws IOException {
this.cleanupJob(jobContext);
}
public void abortJob(JobContext jobContext, int status) throws IOException {
this.cleanupJob(jobContext);
}
public abstract void setupTask(TaskAttemptContext var1) throws IOException;
public abstract boolean needsTaskCommit(TaskAttemptContext var1) throws IOException;
public abstract void commitTask(TaskAttemptContext var1) throws IOException;
public abstract void abortTask(TaskAttemptContext var1) throws IOException;
其中 setupJob在作业运行前被调用，用于初始化操作。当OutputCommitter 被设置为 FileOutputCommitter时，它会创建最终的输出目录${mapreduce.output.fileoutputformat.outputdir}，并为任务的输出创建一个临时文件夹 _temporary，作为最终输出目录的子目录。
FileOutputCommitter 中setupJob() 方法源码如下：
public void setupJob(JobContext context) throws IOException {
if (this.hasOutputPath()) {
Path jobAttemptPath = this.getJobAttemptPath(context);
FileSystem fs = jobAttemptPath.getFileSystem(context.getConfiguration());
if (!fs.mkdirs(jobAttemptPath)) {
LOG.error("Mkdirs failed to create " + jobAttemptPath);
}
} else {
LOG.warn("Output Path is null in setupJob()");
}
}
其中 jobAttemptPath 由 getJobAttemptPath(context) 获取，一层层往下查看此方法调用，最终可以看到FileOutputCommitter 创建的临时目录为：目标输出目录下的_temporary 子目录：
private static Path getPendingJobAttemptsPath(Path out) {
return new Path(out, "_temporary");
}
如果作业成功，则调用 commitJob() 方法。此方法会做临时文件的清理（cleanupJob()），并在最终输出目录中创建名为_SUCCESS的文件，表示Job成功执行完成。若是Job 执行失败，则被状态对象调用abortJob()，默认会调用 cleanupJob() 的方法，对临时文件进行清理。
以上提到的是Job 级别的Committer。在 Task级别，同样也有上述几种方法：
public abstract void setupTask(TaskAttemptContext var1) throws IOException;
public abstract boolean needsTaskCommit(TaskAttemptContext var1) throws IOException;
public abstract void commitTask(TaskAttemptContext var1) throws IOException;
public abstract void abortTask(TaskAttemptContext var1) throws IOException;
其中，在 task 执行之前会调用 setupTask()，但是默认并不做任何工作。因为创建临时任务的输出路径的工作已经在setupJob() 阶段完成。方法needsTaskCommit返回是否需要task 执行提交阶段。提交阶段的工作为：将临时目录下的输出（若有）移动到最终目录。若设置为 false，则执行框架不会为任务运行分布式提交协议，也就不会执行commitTask() 或 abortTask()。当此task没有写任何输出时，FileOutputCommitter会跳过 commit （提交）阶段。
如果task成功执行，并且有输出，则会调用commitTask() 方法，（默认的实现为）将临时目录下的输出文件移动到最终目录（mapreduce.output.fileoutputformat.outputdir）。若是执行失败，则调用abortTask()，删除任务输出的临时目录及文件。
执行框架会保证一个task在有多次尝试的情况下，仅有一个task会被提交。
References: hadoop权威指南第4版