7.4.2　提交任务

任务提交过程是指任务处理完数据后，将最终计算结果从临时目录转移到最终目录的过程。需要注意的是，只有将输出结果直接写到HDFS上的任务才会经历该过程。在Hadoop中，有两种这样的任务：Reduce Task和map-only类型作业的Map Task。

前面提到Hadoop的推测执行机制：Hadoop允许多个任务同时处理同一份数据，但只会选择最先运行完成的任务处理结果作为最终结果。为了防止多个任务产生相同的处理结果而造成冗余，每个任务暂时将自己的计算结果放到一个临时目录中，一旦处理完后，先向JobTracker发送结果提交请求，得到JobTracker准许后，才可以将结果从临时目录转移到最终目录中，而一旦一个任务提交结果后，其计算结果便会作为最终结果，其他任务的计算结果将被丢弃。

Hadoop任务提交过程采用了两阶段提交协议（two-phase commit protocol，2PC）实现。两阶段提交协议是分布式事务中经常采用的协议。它把分布式事务的某一个代理指定为协调者，所有其他代理称为参与者，同时规定只有协调者才有提交或撤销事务的决定权，而其他参与者各自负责在其本地执行写操作，并向协调者提出撤销或提交子事务的意向。两阶段提交协议把事务提交分成两个阶段。

第一阶段（准备阶段）：各个参与者执行完自己的操作后，将状态变为“可以提交”，并向协调者发送“准备提交”请求。

第二阶段（提交阶段）：协调者按照一定原则决定是否允许参与者提交，如果允许，则向参与者发出“确认提交”请求，参与者收到请求后，把“可以提交”状态改为“提交完成”状态，然后返回应答；如果不允许，则向参与者发送“提交失败”请求，参与者收到该请求后，把“可以提交”状态改为“提交失败”状态，然后退出。

对于MapReduce而言，JobTracker扮演协调者的角色，而各个TaskTracker上的任务是参与者。任务提交过程对应的两阶段提交协议实现如图7-4所示。

图　7-4　任务提交序列图

1）Task Attempt处理完最后一条记录后（此时数据写在临时目录${mapred.output.dir}/temporary/${taskid}下），运行状态由RUNNING变为COMMIT_PENDING，并通过RPC将该状态和任务提交请求发送给TaskTracker；

2）TaskTracker得知一个Task Attempt状态为COMMIT_PENDING后，立刻缩短心跳间隔，以便快速将任务状态汇报给JobTracker；

3）JobTracker收到心跳信息后，检查它是否为某个TaskInProgress中第一个状态变为COMMIT_PENDING的Task Attempt，如果是，则批准它进行任务提交，即在心跳应答中添加CommitTaskAction命令，以通知TaskTracker准许该任务提交最终结果；

4）TaskTracker收到CommitTaskAction命令后，将对应任务加入可提交任务列表commitResponses中；

5）Task Attempt通过RPC检测到自己位于列表commitResponses中后，进行结果提交，即将数据从临时目录转移到最终目录（参数${mapred.output.dir}对应的目录）中，并告诉TaskTracker任务提交完成；

6）TaskTracker得知任务提交完成后，将该任务运行状态由COMMIT_PENDING变为SUCCEEDED，并在下次心跳中将该任务状态汇报给JobTracker。

综合以上几个步骤，并结合2PC的定义，很容易知道：上面第1步对应2PC的第一阶段，而其余各步骤对应2PC的第二阶段。