1. org.apache.hadoop.mapreduce.Job#waitForCompletion
/**
* Submit the job to the cluster and wait for it to finish.
* @param verbose print the progress to the user
* @return true if the job succeeded
* @throws IOException thrown if the communication with the
* <code>JobTracker</code> is lost
*/
public boolean waitForCompletion(boolean verbose
) throws IOException, InterruptedException,
ClassNotFoundException {
if (state == JobState.DEFINE) {
// 如果Job.state是DEFINE，就可以提交任务。
submit(); // 参见：2. org.apache.hadoop.mapreduce.Job#submit
}
if (verbose) {
// 在进度和任务失败时实时监控作业和打印状态。
monitorAndPrintJob();
} else {
// get the completion poll interval from the client.
// 获取waitForCompletion() 应检查的时间间隔。
int completionPollIntervalMillis =
Job.getCompletionPollInterval(cluster.getConf());
while (!isComplete()) {
try {
// 任务未完成，则睡眠一会。
Thread.sleep(completionPollIntervalMillis);
} catch (InterruptedException ie) {
}
}
}
// 检查作业是否成功完成。
return isSuccessful();
}
2. org.apache.hadoop.mapreduce.Job#submit
/**
* Submit the job to the cluster and return immediately.
* @throws IOException
*/
public void submit()
throws IOException, InterruptedException, ClassNotFoundException {
// 提交之前，确定Job.state是不是DEFINE，如果不是则抛出异常。
// 因为<设置参数>方法只能在作业提交之前起作用，之后它们将抛出IllegalStateException
ensureState(JobState.DEFINE);
// 默认为新API，除非它们被显式设置，或者使用了旧的mapper或reduce属性。
// 做一些兼容性检查，避免冲突。
// 新API：org.apache.hadoop.mapreduce及其子包
// 老API：org.apache.hadoop.mapred及其子包
setUseNewAPI();
// 初始化org.apache.hadoop.mapreduce.Cluster对象，用于链接/访问map/reduce群集信息。
connect();
// 初始化org.apache.hadoop.mapreduce.JobSubmitter对象。
// 本质就是:new JobSubmitter(FileSystem submitFs, ClientProtocol submitClient)。
// FileSystem ：本地文件系统或者分布式文件系统
// ClientProtocol ：是JobClient和中央JobTracker用于通信的协议。
// JobClient可以使用这些方法提交作业以供执行，并了解当前系统状态。
final JobSubmitter submitter =
getJobSubmitter(cluster.getFileSystem(), cluster.getClient());
// 包含用于向系统提交作业时的一些操作，并返回其最新的Job资料信息。
// 参数：job->要提交的配置；cluster->cluster的句柄(用于链接/访问map/reduce群集信息)
status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
public JobStatus run() throws IOException, InterruptedException,
ClassNotFoundException {
// 参见：《MapReduce源码解析》之二
return submitter.submitJobInternal(Job.this, cluster);
}
});
// 更改Job.state为RUNNING，此时不能再修改任何配置信息。
state = JobState.RUNNING;
LOG.info("The url to track the job: " + getTrackingURL());
}
3. org.apache.hadoop.mapreduce.JobStatus
常用属性：
/**
* Current state of the job
*/
public static enum State {
RUNNING(1),
SUCCEEDED(2),
FAILED(3),
PREP(4),
KILLED(5);
};
// JobID表示作业的不可变且唯一的标识符。
// JobID由两部分组成。
// 第一部分表示jobtracker标识符，以便定义jobID到jobtracker的映射。
// 对于集群设置，此字符串是jobtracker的开始时间，对于本地设置，它是“local”和一个随机数。
// 第二部分是作业编号。
// 示例JobID：job_200707121733_0003，
// 它表示在始于200707121733的jobtracker上运行的第三个作业。
private JobID jobid;
// map进度
private float mapProgress;
// reduce进度
private float reduceProgress;
// 清理进度
private float cleanupProgress;
// 配置进度
private float setupProgress;
// Job的当前状态
private State runState;
// Job的开始时间
private long startTime;
// 提交这个Job的人的userid
private String user;
// 队列名称（任务所属队列，队列指明了任务的优先级）
private String queue;
// Job任务的优先级（VERY_HIGH,HIGH,NORMAL,LOW,VERY_LOW,DEFAULT,UNDEFINED_PRIORITY;）
private JobPriority priority;
// Job的计划信息
private String schedulingInfo="NA";
// Job的失败信息
private String failureInfo = "NA";
// Job的权限控制信息，有2个内置的枚举对象：
// org.apache.hadoop.mapreduce.JobACL#VIEW_JOB
// org.apache.hadoop.mapreduce.JobACL#MODIFY_JOB
private Map<JobACL, AccessControlList> jobACLs =
new HashMap<JobACL, AccessControlList>();
// Job名称
private String jobName;
// Job配置文件
private String jobFile;
// Job的完成时间
private long finishTime;
// 用于检查作业状态是否已经被标记为“retired”。
// 在Hadoop中，当作业完成或失败时，它可能会被标记为“retired”，这意味着它不再是运行作业列表中的活动作业。
private boolean isRetired;
// 已完成作业的历史文件。若作业未完成或历史文件不可用，则为null。
private String historyFile = "";
// 链接到web用户界面的url
private String trackingUrl ="";
// 已使用slot数量
// slot不是CPU的Core，也不是memory chip，它是一个逻辑概念.
// 一个节点的slot的数量用来表示某个节点的资源的容量或者说是能力的大小，因而slot是Hadoop的资源单位。
private int numUsedSlots;
// 剩余slot数量
private int numReservedSlots;
// 已使用内存
private int usedMem;
// 保留内存
private int reservedMem;
// 所需内存
private int neededMem;
// Job是否运行在Uber模式中
// uber模式是 `2.x` 开始引入的；
// 以 `Uber` 模式运行 MR 作业，所有的 `Map Tasks` 和 `Reduce Tasks` 将会在 `ApplicationMaster` 所在的容器（`container`）中运行；
// 即：整个 MR 作业运行的过程 只会 启动 `AM container`，所有子task 与 ApplicationMaster 在同一个JVM中执行，达到JVM重用的目的；
// 因为不需要启动 `mapper containers` 和 `reducer containers`，所以 AM 不需要 和远程 containers 通信，整个过程简单，执行速度快；
// 如果 MR 作业 输入的数据量 非常小，启动 `Map container` 或 `Reduce container` 的时间都 比处理数据要长，那么这个作业就可以考虑启用 `Uber` 模式运行；
// 一般情况下，对小作业启用 Uber 模式运行会得到 `2-3`倍 的性能提升。
private boolean isUber;
