3.6 Spark RDD容错原理及其四大核心要点解析
本节讲解RDD不同的依赖关系（宽依赖、窄依赖）的Spark RDD容错处理；对Spark框架层面容错机制的三大层面（调度层、RDD血统层、Checkpoint层）及Spark RDD容错四大核心要点进行深入解析。
3.6.1 Spark RDD容错原理
RDD不同的依赖关系导致Spark对不同的依赖关系有不同的处理方式。
对于宽依赖而言，由于宽依赖实质是指父RDD的一个分区会对应一个子RDD的多个分区，在此情况下出现部分计算结果丢失，单一计算丢失的数据无法达到效果，便采用重新计算该步骤中的所有数据，从而会导致计算数据重复；对于窄依赖而言，由于窄依赖实质是指父RDD的分区最多被一个子RDD使用，在此情况下出现部分计算的错误，由于计算结果的数据只与依赖的父RDD的相关数据有关，所以不需要重新计算所有数据，只重新计算出错部分的数据即可。
3.6.2 RDD容错的四大核心要点
Spark框架层面的容错机制，主要分为三大层面（调度层、RDD血统层、Checkpoint层），在这三大层面中包括Spark RDD容错四大核心要点。
Stage输出失败，上层调度器DAGScheduler重试。
Spark计算中，Task内部任务失败，底层调度器重试。
RDD Lineage血统中窄依赖、宽依赖计算。
Checkpoint缓存。
1．调度层（包含DAG生成和Task重算两大核心）
从调度层面讲，错误主要出现在两个方面，分别是在Stage输出时出错和在计算时出错。
1）DAG生成层
Stage输出失败，上层调度器DAGScheduler会进行重试，如下列源码所示。
DAGScheduler.scala的resubmitFailedStages的源码如下。
1. private[scheduler] def resubmitFailedStages() {
2. //判断是否存在失败的Stages
3. if (failedStages.size > 0) {
4. //失败的阶段可以通过作业取消删除，如果ResubmitFailedStages事件已调度，失
//败将是空值
5.
6. logInfo("Resubmitting failed stages")
7. clearCacheLocs()
8. //获取所有失败Stage的列表
9. val failedStagesCopy = failedStages.toArray
10. //清空failedStages
11. failedStages.clear()
12. //对之前获取的所有失败的Stage，根据jobId排序后逐一重试
13. for (stage <- failedStagesCopy.sortBy(_.firstJobId)) {
14. submitStage(stage)
15. }
16. }
17. }
2）Task计算层
Spark计算过程中，计算内部某个Task任务出现失败，底层调度器会对此Task进行若干次重试（默认4次）。
TaskSetManager.scala的handleFailedTask的源码如下。
1. def handleFailedTask(tid: Long, state: TaskState, reason:
TaskFailedReason) {
2. ......
3. if (!isZombie && reason.countTowardsTaskFailures) {
4. taskSetBlacklistHelperOpt.foreach(_.updateBlacklistForFailedTask(
5. info.host, info.executorId, index))
6. assert (null != failureReason)
7. //对失败的Task的numFailures进行计数加一
8. numFailures(index) += 1
9. //判断失败的Task计数是否大于设定的最大失败次数，如果大于，则输出日志，并不再重试
10. if (numFailures(index) >= maxTaskFailures) {
11. logError("Task %d in stage %s failed %d times; aborting job".format(
12. index, taskSet.id, maxTaskFailures))
13. abort("Task %d in stage %s failed %d times, most recent failure:
%s\nDriver stacktrace:"
14. .format(index, taskSet.id, maxTaskFailures, failureReason),
failureException)
15. return
16. }
17. }
18. //如果运行的Task为0时，则完成Task步骤
19. maybeFinishTaskSet()
20. }
21. .......
2．RDD Lineage血统层容错
Spark中RDD采用高度受限的分布式共享内存，且新的RDD的产生只能够通过其他RDD上的批量操作来创建，依赖于以RDD的Lineage为核心的容错处理，在迭代计算方面比Hadoop快20多倍，同时还可以在5～7s内交互式地查询TB级别的数据集。
Spark RDD实现基于Lineage的容错机制，基于RDD的各项transformation构成了compute chain，在部分计算结果丢失的时候可以根据Lineage重新恢复计算。
在窄依赖中，在子RDD的分区丢失，要重算父RDD分区时，父RDD相应分区的所有数据都是子RDD分区的数据，并不存在冗余计算。
在宽依赖情况下，丢失一个子RDD分区，重算的每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区用的，会有一部分数据相当于对应的是未丢失的子RDD分区中需要的数据，这样就会产生冗余计算开销和巨大的性能浪费。
3．checkpoint层容错
Spark checkpoint通过将RDD写入Disk作检查点，是Spark lineage容错的辅助，lineage过长会造成容错成本过高，这时在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的RDD开始重做Lineage，就会减少开销。
checkpoint主要适用于以下两种情况：
DAG中的Lineage过长，如果重算，开销太大，如PageRank、ALS等。
尤其适合于在宽依赖上作checkpoint，这个时候就可以避免为Lineage重新计算而带来的冗余计算。