首页 > 代码库 > 大数据：Spark Standalone 集群调度（三）多Master节点的可用性

大数据：Spark Standalone 集群调度（三）多Master节点的可用性

2024-09-11 21:29:53 221人阅读

1. Master 单节点可用性

Master节点在Spark中所承载的作用是分配Application到Worker节点，维护Worker节点，Driver，Application的状态。

在Spark中，Master本身也提供了基于硬盘的单节点的可用性，也就是可以直接通过重启Master，Master通过读取硬盘里保存的状态，进行单节点的恢复。

In order to enable this recovery mode, you can set SPARK_DAEMON_JAVA_OPTS in spark-env using this configuration:

System property	Meaning
`spark.deploy.recoveryMode`	Set to FILESYSTEM to enable single-node recovery mode (default: NONE).
`spark.deploy.recoveryDirectory`	The directory in which Spark will store recovery state, accessible from the Master‘s perspective.

通过设置recoveryMode为FILESYSTEM，当Master节点Down掉时候，节点的可用性通过重启Master节点来恢复，Master会通过获取磁盘里保存的集群原来的状态进行恢复。

如果没有设置recoveryMode，默认为不支持恢复，Master重启的时候，不会进行对前状态进行恢复。

备注：恢复节点的方式与多节点可用性后续的处理方式基本类似，就不在此具体讨论细节了

2. Master 多节点可用性

单节点的问题很明显，虽然有恢复的方案，很明显在重启的过程中，是无法继续对资源调度和分配的，而磁盘也是具有单节点特性（假使没有构建磁盘阵列），同样也不具备高可用性的特点

Spark引入了多Master节点来提高集群可用性

设置多Master节点的方式:

-Dspark.deploy.recoveryMode=ZOOKEEPER

2.1 StandBy Master 节点

在Spark的多Master节点设计中，分为两个角色，一个是主Master节点提供服务，另一个角色是备用Master节点

当Master节点处于备用的状态时候，是不提供服务的，备用节点只是在处于等待状态，只有当主Master节点crash掉后，备用节点才升级成为主Master提供服务

主备节点互相不感知，不提供负载均衡，不是对等节点，在上图可以看到每个提交任务的Client并不提交任务到StandBy节点

2.2 ZooKeeper 选举Main Master节点

在Master多节点中，只有Main Master才是提供服务的，这个设计就象选举Leader，这种方式非常适合使用Zookeeper的Watch模式。

Spark并没有直接使用Zookeeper，而是使用了curator 框架（基于Zookeeper）进行Master的选举。由于Zookeeper的操作比较底层，操作的是数据结构和节点的状态，curator在上面封装了一层，以应用与不同的应用情景。

在使用Leader模式下，curator的框架开发代码非常简单。使用LeaderLatch, 实现LeaderLatchListener的监听器

override def isLeader() {
    synchronized {
      // could have lost leadership by now.
      if (!leaderLatch.hasLeadership) {
        return
      }

      logInfo("We have gained leadership")
      updateLeadershipStatus(true)
    }
  }

  override def notLeader() {
    synchronized {
      // could have gained leadership by now.
      if (leaderLatch.hasLeadership) {
        return
      }

      logInfo("We have lost leadership")
      updateLeadershipStatus(false)
    }
  }

isLeader 和 notLeader 分别代表成为Leader或者失去Leader，在Spark的ZookeeperLeaderElectionAgent.scala里跟新leader的状态

  private def updateLeadershipStatus(isLeader: Boolean) {
    if (isLeader && status == LeadershipStatus.NOT_LEADER) {
      status = LeadershipStatus.LEADER
      masterInstance.electedLeader()
    } else if (!isLeader && status == LeadershipStatus.LEADER) {
      status = LeadershipStatus.NOT_LEADER
      masterInstance.revokedLeadership()
    }
  }

调用的是在Master.scala的

  override def electedLeader() {
    self.send(ElectedLeader)
  }

  override def revokedLeadership() {
    self.send(RevokedLeadership)
  }

也就是发了消息ElectedLeader, RevokeLeaderShip 给自己的Netty

Zookeeper默认的leader节点 /spark/leader_election

ls /spark/leader_election
[_c_6b494d00-cf91-4d69-8828-942377bafdf1-latch-0000000217, _c_c1f4c10f-9479-4a55-b660-0497c0b54f89-latch-0000000218]

前面应该是UUID后面有一个序列号，由后面的序列号前后来决定谁是 leader，而序列号前后的生成取决于Master连接zookeeper的顺序，熟悉zookeeper的知道这些节点是ephemeral节点。

2.3 StandBy Master 节点的恢复

在前面章节里已经描述了如何选为Main Master，当Main Master 当机后，备用的Master通过选举成为了 Main Master，备用的Master Netty收到了ElectedLeader的消息，在成为主Master节点前，先检查集群状态并准备进行恢复

    case ElectedLeader =>
      val (storedApps, storedDrivers, storedWorkers) = persistenceEngine.readPersistedData(rpcEnv)
      state = if (storedApps.isEmpty && storedDrivers.isEmpty && storedWorkers.isEmpty) {
        RecoveryState.ALIVE
      } else {
        RecoveryState.RECOVERING
      }
      logInfo("I have been elected leader! New state: " + state)
      if (state == RecoveryState.RECOVERING) {
        beginRecovery(storedApps, storedDrivers, storedWorkers)
        recoveryCompletionTask = forwardMessageThread.schedule(new Runnable {
          override def run(): Unit = Utils.tryLogNonFatalError {
            self.send(CompleteRecovery)
          }
        }, WORKER_TIMEOUT_MS, TimeUnit.MILLISECONDS)
      }

2.3.1 获取具体的信息

首先从持久化层获取Application, Drivers, Worker 的信息，也就是通过Curator 的框架获取原Master保存的信息

Zookeeper默认的master状态 /spark/master_status

 ls /spark/master_status  
[worker_worker-20170329184022-192.168.121.101-55109, worker_worker-20170330035822-192.168.121.102-41307]

Apps, Driver, Workers 的信息都会被保存在master_status节点里，而每个子节点的内容就是序列化过的ApplicationInfo, DriverInfo, WorkInfo

获选的主Master通过获取Zookeeper上的节点的值，反序列化重构Application, Driver, Worker的信息

2.3.2 恢复Application,Driver,Workers

private def beginRecovery(storedApps: Seq[ApplicationInfo], storedDrivers: Seq[DriverInfo],
      storedWorkers: Seq[WorkerInfo]) {
    for (app <- storedApps) {
      logInfo("Trying to recover app: " + app.id)
      try {
        registerApplication(app)
        app.state = ApplicationState.UNKNOWN
        app.driver.send(MasterChanged(self, masterWebUiUrl))
      } catch {
        case e: Exception => logInfo("App " + app.id + " had exception on reconnect")
      }
    }

    for (driver <- storedDrivers) {
      // Here we just read in the list of drivers. Any drivers associated with now-lost workers
      // will be re-launched when we detect that the worker is missing.
      drivers += driver
    }

    for (worker <- storedWorkers) {
      logInfo("Trying to recover worker: " + worker.id)
      try {
        registerWorker(worker)
        worker.state = WorkerState.UNKNOWN
        worker.endpoint.send(MasterChanged(self, masterWebUiUrl))
      } catch {
        case e: Exception => logInfo("Worker " + worker.id + " had exception on reconnect")
      }
    }
  }

2.3.2.1 恢复Application

在前面一章说过如何分配Executor，里面也说到了ApplicationInfo 结构。在Master做Recovery的时候，把保存在Zookeeper里的Applications 重新在Master里重新注册

这里有几个注意点：

1. Zookeeper 里保存的是正在运行的Applications，但在Spark里并没有进行细粒度的控制，只要是在运行的Applications，都会重新进行被分配executor，哪怕正在分配或者已经在计算了。

2. 注册完Application 并没有马上进行Application的调度，直到恢复完Worker，Driver才开始进行进行

  private def completeRecovery() {
    // Ensure "only-once" recovery semantics using a short synchronization period.
    if (state != RecoveryState.RECOVERING) { return }
    state = RecoveryState.COMPLETING_RECOVERY

    // Kill off any workers and apps that didn‘t respond to us.
    workers.filter(_.state == WorkerState.UNKNOWN).foreach(removeWorker)
    apps.filter(_.state == ApplicationState.UNKNOWN).foreach(finishApplication)

    // Reschedule drivers which were not claimed by any workers
    drivers.filter(_.worker.isEmpty).foreach { d =>
      logWarning(s"Driver ${d.id} was not found after master recovery")
      if (d.desc.supervise) {
        logWarning(s"Re-launching ${d.id}")
        relaunchDriver(d)
      } else {
        removeDriver(d.id, DriverState.ERROR, None)
        logWarning(s"Did not re-launch ${d.id} because it was not supervised")
      }
    }

    state = RecoveryState.ALIVE
    schedule()
    logInfo("Recovery complete - resuming operations!")
  }

在completeRecovery 的函数里，只有当Master的节点的状态RecoveryState.ALIVE的时候，Master才调用了Schedule重新调度已经注册的Application

2.3.2.2 恢复Worker

第一步：注册Worker，这里只是简单的确认了Worker地址（ip+port）的格式是否正确和是否已经注册过了。

第二步：设置Worker的状态为UNKNOW

第三步：发送MasterChange消息给Worker

第四步：Worker将自己的Application ID, Executor的信息WorkerSchedulerStateResponse发回给Master

第五步：Master的跟新信息中ApplicationInfo中的Executor信息

第三步到第五步都是通过消息来异步消息恢复的，在没有收到WorkerScheduleStateResponse，是无法开始准确进行Application调度的，此时的Worker的里的信息不准确，例如用了多少核

2.3.3 完成Recovery

因为要完成worker信息的传递，而Recovery的过程是在Netty的接收ElectedLeader消息的线程中进行的，而比如象Worker汇报消息的不能被堵塞，在这里Master起了一个定时任务去触发任务的完成，定时任务的时间就是worker连接的超时时间，在这里的前提是所有的workers会在超时间段内回复消息。

recoveryCompletionTask = forwardMessageThread.schedule(new Runnable {
          override def run(): Unit = Utils.tryLogNonFatalError {
            self.send(CompleteRecovery)
          }
        }, WORKER_TIMEOUT_MS, TimeUnit.MILLISECONDS)

既然还是发消息，对Master发送了CompleteRecovery的消息，在CompleteRecovery的消息中重新调度了Application。

2.3.4 Recovery中的几个状态

Master节点在recovery中有四个状态：

RecoveryState.ALIVE
RecoveryState.RECOVERING
RecoveryState.COMPLETING_RECOVERY
RecoveryState.STANDBY

ALIVE ----- 活着

RECOVERING ------ 恢复的过程中

COMPLETING_RECOVERY ----------运行在completeRecovery函数中

STANDBY ------------ 备份节点

在整个Recovery的恢复过程中状态是从 STANDBY => RECOVERING => COMPLETING_RECOVERY => ALIVE

1. 在整个Recovery的恢复过程中可以Register Application，当然必须Elected_Leader的消息函数块运行完，因为在同一个线程中

2. 在整个Recovery的恢复过程中只有在进入ALIVE的状态，才能进行Application的调度

3 Submit 如何设置多个Master

Spark的master是主和Standby的状态，对submit任务来说，并不能知道目前集群上主Master是那个，提交到Standby的master并不接受任务，容易导致提交任务失败。

Spark 提供了一种简单的方式：在submit 或者在sparkconf里可以设置多个Master, 格式注意是用逗号分隔

--master spark://raintungmaster:7077,raintungslave1:7077

当然你可以也可以设置一个局域网的DNS,通过 healthcheck来判断哪个Master挂了，DNS就指向谁

大数据：Spark Standalone 集群调度（三）多Master节点的可用性

声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉：投诉/举报工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。

联系
我们

首页 > 代码库 > 大数据：Spark Standalone 集群调度（三）多Master节点的可用性

大数据：Spark Standalone 集群调度（三）多Master节点的可用性

1. Master 单节点可用性

2. Master 多节点可用性

2.1 StandBy Master 节点

2.2 ZooKeeper 选举Main Master节点

2.3 StandBy Master 节点的恢复

2.3.1 获取具体的信息

2.3.2 恢复Application,Driver,Workers

2.3.2.1 恢复Application

2.3.2.2 恢复Worker

2.3.3 完成Recovery

2.3.4 Recovery中的几个状态

3 Submit 如何设置多个Master

看完仍有疑问？有类似问题直接问程序猿