Spark 的 Failover 机制全解析 (Master/Worker/Executor)

Posted 2021-04-27 数据社

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Spark 的 Failover 机制全解析 (Master/Worker/Executor)相关的知识，希望对你有一定的参考价值。

后台回复【加群】，申请加入数据学习交流群

所谓容错是指一个系统的部分出现错误的情况还能够持续地提供服务，不会因为一些细微的错误导致系统性能严重下降或者出现系统瘫痪。在一个集群出现机器故障、网络问题等是常态，尤其集群达到较大规模后，很可能较频繁出现机器故障不能进行提供服务，因此对于分布式集群需要进行容错设计。Spark在设计之初考虑到这种情况，所以它能够实现高容错，以下将从ExecutorWorker和Master的异常处理来介绍。

Executor 异常

Spark支持多种运行模式，这些运行模式中的集群管理器会为任务分配运行资源，在运行资源中启动Executor,由Executor是负责执行任务的运行，最终把任务运行状态发送给Driver。

下面将以独立运行模式分析Executor出现异常的情况，其运行结构如图所示，其中虚线为正常运行中进行消息通信线路，实线为异常处理步骤。

Spark 的 Failover 机制全解析 (Master/Worker/Executor)

1. 首先看 Executor 的启动过程：在集群中由 Master 给应用程序分配运行资源后，然后在 Worker中启动 ExecutorRunner ,而 ExecutorRunner 根据当前的运行模式启动 CoarseGrainedExecutorBackend 进程，当该进程会向Driver发送注册Executor信息，如果注册成功，则 CoarseGrainedExecutorBackend 在其内部启动 Executor。Executor 由 ExecutorRunner 进行管理，当Executor出现异常时(如所运行容器 CoarseGrainedExecutorBackend 进程异常退出等)，由 ExecutorRunner 捕获该异常并发送 ExecutorStateChanged 消息给 Worker。

2. Worker 接收到 ExecutorStateChanged 消息时，在 Worker 的 handleExecutorStateChanged 方法中，根据Executor状态进行信息更新，同时把Executor状态信息转发给Master。

3. Master接收到Executor状态变化消息后，如果发现Executor出现异常退出，则调用 Master.schedule 方法，尝试获取可用的 Worker 节点并启动 Executor,而这个 Worker 很可能不是失败之前运行Executor的Worker节点。该尝试系统会进行10次，如果超过10次，则标记该应用运行失败并移除集群中移除该应用。这种限定失败次数是为了避免提交的应用程序存在 Bug 而反复提交，进而挤占集群宝贵的资源。

Worker 异常

Spark独立运行模式采用的是Master/Slave的结构，其中Slave是有Worker来担任的，在运行的时候会发送心跳给Master,让Master知道Worker的实时状态,另一方面Master也会检测注册的Worker是否超时，因为在集群运行过程中，可能由于机器宕机或者进程被杀死等原因造成Worker进程异常退出。下面将分析Spark集群如何处理这种情况，其处理流程如图所示。

1. 这里需要了解Master是如何感知到Worker超时？在Master接收Worker心跳的同时，在其启动方法onStart中启动检测Worker超时的线程，其代码如下：

      
        
        
      
       
         
         
       checkForWorkerTimeOutTask = forwardMessageThread. scheduleAtFixedRate (new Runnable { 
       
         
         
       override def run (): Unit = Utils.tryLogNonFatalError (
       
         
         
       //非自身发送消息CheckForWorkerTimeOut,调用timeOutDeadWorkers方法进行检测
       
         
         
       self.send(CheckForWorkerTimeOut)
       
         
         
       }
       
         
         
       }, 0, WORKER_TIMEOUT_MS, TimeUnit.MILLISECONDS)

2. 当Worker出现超时时，Master调用timeOutDeadWorkers方法进行处理，在处理时根据Worker运行的是Executor和Driver分别进行处理。

如果是 Executor, Master 先把该 Worker 上运行的 Executor 发送消息 ExecutorUpdated 给对应的 Driver,告知 Executor 已经丢失，同时把这些 Executor 从其应用程序运行列表中删除。另外，相关Executor的异常也需要按照前一小节进行处理。
如果是Driver,则判断是否设置重新启动。如果需要，则调用Master.schedule方法进行调度，分配合适节点重启Driver；如果不需要重启，则删除该应用程序。

Master 异常

Master作为Spark独立运行模式中的核心，如果Master出现异常，则整个集群的运行情况和资源将无法进行管理，整个集群将处于“群龙无首”的状况。很幸运的是，Spark在设计时考虑了这种情况，在集群运行的时候，Master将启动一个或多个Standby Master,当 Master 出现异常的时候，Standby Master 将根据一定规则确定其中一个接管 Master。

在独立运行模式中， Spark支持如下几种策略，可以在配置文件spark-env.sh配置项spark .deploy, recovery Mode进行设置，默认为NONE。

ZOOKEEPER：集群的元数据持久化到ZooKeeper中，当Master出现异常时.ZooKeeper 会通过选举机制选举出新的Master,新的Master接管时需要从ZooKeeper获取持久化信息并根据这些信息恢复集群状态。具体结构如图4-13所示。
FILESYSTEM：集群的元数据持久化到本地文件系统中，当Master出现异常时，只要在该机器上重新启动Master,启动后新的Master获取持久化信息并根据这些信息恢复集群状态。
CUSTOM：自定义恢复方式，对StandaloneRecoveryModeFactory抽象类进行实现并把该类配置到系统中，当Master出现异常时，会根据用户自定义的方式进行恢复集群状态。
NONE：不持久化集群的元数据，当Master出现异常时，新启动的Master不进行恢复集群状态，而是直接接管集群。