容错机制java_Spark 容错机制

Spark 容错机制

任何容错机制的设计都是先考虑正常情况下是如何处理的，然后去考虑各种失败场景，失败场景可分 Crash(kill -9，掉电等)，正常退出(例如抛异常，程序可以做善后处理)，网络分区。

Task

我们先考虑最底层的失败，即某一个 Task 执行失败了。

先来看应该如何处理：

某 task A 因为取 shuffle 数据取失败而失败了。

首先，确认失败前应该重试几次，以防止网络分区造成的短暂失败

然后应该和 driver 通信，driver 负责重新跑该数据对应的 ShuffleMapTask B，driver 需要记录 A 在等待 B 的数据，并且当 B 好了之后通知 A，而 A 与 driver 通信后就睡觉，等待 B 好了之后 driver 的唤醒。

不能取消 A，也不能认为 A 对应的 Stage 失败。不能取消 A 是因为不能浪费 A 已经完成的工作，例如 A 可能已经取了别的 shuffle 数据，不能因为一个 ShuffleMapTask 的输出丢了就丢掉之前取的数据。对于 A 对应的 Stage 可以这样处理：还未启动的 task 可以先不启动；已经启动的 task 之后可能会遇到取失败，如果遇到，则像 A 一样处理就行；已经启动的 task 可能已经取完那部分数据了，因此最后可能成功。

其他原因的失败只涉及 task 内部，因此重启该 task 就行。

上面是我能想到的最优的处理，我们来看 spark 是如何做的。

Task 失败由 executor 感知到，通过 statusUpdate 层层传递到 driver 端的 TaskSetManager.handleFailedTask，其基本逻辑是：

如果是 FetchFailed，即 reduce task 拿不到 shuffle 数据，那么上一个 Stage 需要重跑。(FetchFailed 的处理之后会详讲)

如果不是 FetchFailed，并且该 Task 的其他尝试(spark 可能会启动 task 的多个副本)还未成功，会重新启动该 task。

当然，重启次数超过 spark.task.maxFailures，taskSet 会失败，这意味着一个 stage 失败了。stage 失败整个任务就失败了，spark 会取消该 stage 对应的 job 包含的所有 task，并返回用户任务执行失败。

FetchFailed

我们来考虑 FetchFailed 的处理，值得注意的是 FetchFailed 的处理会在未来有比较大的变化，见 SPARK-20178。

FetchFailed 会在 ShuffleReader 取数据失败 N 次后抛出，然后由 executor 通过 statusUpdate 传到 driver 端，实际的处理会在 DAGScheduler.handleTaskCompletion，它会重新提交该 Stage 和该 Stage 对应的 ShuffleMapStage，重试次数超过 spark.stage.maxConsecutiveAttempts 时会退出。

Speculative task

有时候 task 不是失败了，而是太慢了，这时 spark 会多启动几个实例(spark 中一个实例叫 attempt)，有任何一个实例成功了该 task 就成功了。spark 会周期性检查是否有 task 符合 speculation 的条件，检查逻辑在 TaskSetManager.checkSpeculatableTasks。

那么多个实例如何防止其输出不干扰呢？例如一个实例 A成功了，输出，另一个实例 B成功了，其输出覆盖了 A 的输出，甚至更糟糕的是两者同时写一个文件，导致输出错乱。

有几个层面：

首先，输出到文件是通过 ShuffleWriter.write 写的。以 SortShuffleWriter 为例，最后在 IndexShuffleBlockResolver.writeIndexFileAndCommit 里会先写数据到临时文件，然后检查文件是否存在，如果文件不存在，就重命名临时文件为正式文件。

如果两个 task attempt 在同一个 executor 执行，这种情况是不可能的，调度时保证两个 attempt 不会运行在同一台机器上，详情见 TaskSetManager.dequeueSpeculativeTask(IndexShuffleBlockResolver中，检查文件是否存在和文件重命名是在一个 synchronized 块里的，这个 synchronized 看起来没什么必要)

如果两个 task attempt 在两个 executor 执行，但这两个 executor 在同一台机器上。这种情况在 speculative task 是不可能出现的，这在 speculative task 调度时保证。(不过好像也没事，就是重命名文件时，后面的覆盖前面的，反正内容都一样)

两个 task attempt 在不同机器上运行，这时两台机器上都有输出文件

一个 attempt 成功了，会在 TaskSetManager.handleSuccessfulTask 中取消其他正在运行的 task 实例。如果在取消前，有另一个 attempt 成功了，那么另一个 attempt 的输出会注册到 mapOutputTracker 里，从而覆盖之前的实例的注册，这会导致不同 task 会读不同位置的输出，可能会有问题。

Executor

Executor crash

executor crash 是指 executor 异常退出了，比如 executor JVM 崩溃了。

先来看应该如何处理。

可以通过心跳，RPC 通信失败，外部的 executor 管理程序来检测 executor

executor 所运行的 task 应该重新被调度到其他 executor

如果没有 external shuffle server，那么 shuffle 数据也丢了，需要重新启动生成 shuffle 数据的 task

重启 executor

我们来看 spark 是如何做的。在 deploy mode 为 spark standalone 情况下，系统实际运行的进程是 CoarseGrainedExecutorBackend(粗粒度的excutor监控，比如Executor的启停等)，所以 Eexecutor crash 就等于该进程 crash。有几个检测 crash 的方式：

CoarseGrainedSchedulerBackend.onDisconnected 这是 RPC 系统发现 executor 连不上了，该函数会调用 removeExecutor，进而调用 TaskSchedulerImpl.removeExecutor。

Worker 里的 ExecutorRunner 会监控 executor 进程，其结束时会发送 ExecutorStateChanged 给 Worker，从而到达 Master，最后到达 CoarseGrainedSchedulerBackend.removeExecutor。Master 会重启 executor，一个 app 的 executor 重启超过 spark.deploy.maxExecutorRetries，app 会被终止。

sparkContext 里有个 HeartbeatReceiver，如果一段时间内没收到 executor 的心跳，会触发 TaskSchedulerImpl.executorLost。该函数也会在 CoarseGrainedSchedulerBackend.removeExecutor 中被调用。

如果 deploy mode 为 mesos，如果运行模式是 Coarse-Grained(Fine-Grained 已经是 deprecated 的了)，实际运行的也是 CoarseGrainedExecutorBackend。在该情况下 executor crash 会通过 MesosCoarseGrainedSchedulerBackend.statusUpdate 处理，它会从totalCoresAcquired 中减去已使用的 cpu，从而在下次调度时，由于 totalCoresAcquired < spark.cores.max，会启动一个新的 executor，没有重启次数限制。

executor 挂了，task 会在 TaskSchedulerImpl.removeExecutor 里被标记为失败，而 task 失败的处理上一节介绍过了。

Executor 网络分区

Executor 网络分区指的是 executor 没有挂，但是和 driver 的网络连接断了，之后可能会恢复。

先来应该如何处理。系统是无法区分网络分区和 crash 的，因此，系统会把网络分区当做 Crash 处理，关键是在 executor 重新连上时该如何做。重新连上时，driver 端应该忽略， executor 发出的任何消息，并且通知 executor 自杀。

我们来看 spark 如何做的。executor 是如何自杀的呢：

如果 executor 向 driver 发送 Heartbeat 的失败次数超过 spark.executor.heartbeat.maxFailures，executor 会自杀，默认设置下，10 分钟连不上，executor 就自杀了，所以这个机制起的是保底的作用，防止在任何情况下 executor 的泄露。

CoarseGrainedExecutorBackend.onDisconnected 会杀掉 executor

driver 端 CoarseGrainedSchedulerBackend 在清理 executor 状态时会向 executor 发送 StopExecutor

driver 如何忽略旧 executor 发出的消息：

现在的实现里，driver 并没有忽略旧 executor 发出的消息，这可能会造成问题，可以得到改进。具体来说，在 CoarseGainedSchedulerBackend.receive StatusUpdate，并没有看 executorId 是否依旧在 driver 的表中。

Driver

Driver crash

还是先看应该如何做。

如果想尽量保留 driver 已完成的工作，driver 挂了，应该依赖持久化存储和与 executor 的通信，恢复 driver 的状态，重现建立对正在运行的 executor，task 等的管理。而 executor 与 driver 通信失败时应该有重试机制，并且当重试失败时不应该退出，而应该等待 driver 恢复后重连，并把自己的状态告诉 driver，只有在超过 driver 恢复时间时才自行退出。

spark 的实现。不管什么模式，driver 挂了，所有相关的 executor 和 task 都会被清理，不尝试恢复，在 yarn-cluster 模式下还会重启 driver，重跑所有 task。这种方式浪费了已有的工作，但实现起来是最简单的。

Driver 网络分区

应该忽略旧 driver 的所有消息，并且让旧 driver 退出。

在 spark 的实现下，例如在 yarn-cluster 模式，如果 driver 网络分区，yarn 会重启 driver，旧 driver 重连时，可能会有点问题，因为两个 driver 的 ID 是一样的，spark 分不清两者。

主机

主机 Crash

主机 Crash，只需要把主机上运行的 driver，executor，task 都按失败处理，同时上面的 shuffle 数据对应的 task 也得重跑。driver 等失败的处理已经介绍过了。

主机网络分区

按主机 Crash 处理，上面的旧 driver 的处理也按 driver 网络分区一样处理，其他对象以此类推。

总结

任何容错机制的设计都是先考虑正常情况下是如何处理的，然后去考虑各种失败场景，失败场景可分 Crash(kill -9，掉电等)，正常退出(例如抛异常，程序可以做善后处理)，网络分区。本文介绍了 spark 的 driver，executor，task 失败时的处理，同时也对各种情况应该如何处理表达了一些看法，希望有所帮助。