使用Hadoop来运行你的作业的其中一个主要原因就是它的高容错性,就算在由高失败率的节点或网络组成的大集群内运行的作业,Hadoop都可以让作业成功完成。
Hadoop实现容错的主要方法就是重新执行任务,单个任务节点(TaskTracker)会不断的与系统的核心节点(JobTracker)进行通信,如果一个TaskTracker在一定时间内(默认是1分钟)无法与JobTracker进行通信,那JobTracker会假设这个TaskTracker出问题挂了,JobTracker了解给每个TaskTracker赋予了那些map和reduce任务。
如果作业仍然在mapping阶段,其它的TaskTracker会被要求重新执行所有的由前一个失败的TaskTracker所执行的map任务。
如果作业在reduce阶段,则其它的TaskTracker会被要求重新执行所有的由前一个失败的TaskTracker所执行的reduce任务。
Reduce任务一旦完成会把数据写到HDFS。
因此,如果一个TaskTracker已经完成赋予它的3个reduce任务中的2个,那只有第三个任务会被重新执行。Map任务则更复杂一点:即使一个节点已经完成了10个map任务,reducer仍可能无法获取这些map任务的所有的输出。如果此时节点挂了,那它的mapper输出就不可访问了。所以已经完成的map任务也必须被重新执行以使它们的输出结果对剩下的reducing机器可用,所有的这些都是由Hadoop平台自动操作完成的。
这个容错性强调需要程序的执行没有副作用影响,如果Mapper和Reducer有自身的标识并和外部有通信,那重新执行一个任务可能需要其它节点去和新的map或reduce任务实例进行通信,并且重启的任务可能需要重建它们的中间状态。
这个过程是很复杂的并且容易出错。MapReduce通过去除任务标识或任务间的通信而大大简化了这个问题。单个任务只能看到它自己的输入和输出,这样就使得错误与重启过程变成清晰可靠。
推测性的执行(Speculative execution):Hadoop系统有一个问题,它把任务分派到很多个节点,其中很有可能有一些慢的节点会限制剩下程序的执行速度。
举个例子,如果有个节点内有一个比较慢的磁盘控制器,那它读取输入数据的速度可能只有所有其它节点的速度的10%。
所以当99个map任务都已经完成了,系统仍在等待最后那个比较耗时的map任务完成。
通过强迫任务独立运行于其它的任务,使得单个任务之间不会知道它们的输入数据来自哪里。
任务相信Hadoop平台会派送合适的输入到它们那里。因此,对于相同的输入数据,我们可以并行多次处理以利用不同机器的负载能力。
因为作业中大多数的任务都已经完成了,
Hadoop平台会在几个空闲的节点上调度执行剩余任务的拷贝,这个过程叫做推测性的执行。
当任务完成时,它会向JobTracker通告。任何一个首先完成的拷贝任务将成为权威拷贝,如果其他拷贝任务还在推测性的执行中,Hadoop会告诉TaskTracker去终止这些任务并丢弃它们的输出,接着Reducer会从首先完成的Mapper那里获取输入数据。
推测性的执行默认是启用的,你可以通过设置JobConf中的mapred.map.tasks.speculative.execution和mapred.reduce.tasks.speculative.execution为false来禁用mapper和reducer的推测性的执行。