Spark源码之Standalone模式下master持久化引擎讲解

      

      Standalone 模式下Master为了保证故障恢复，会持久化一些重要的数据，来避免master故障导致集群不可用这种情况(也即单点故障)。目前，有四种持久化策略:

1、基于zookeeper的持久化引擎；

2、基于文件的持久化引擎；

3、用户自定义持久化引擎；

4、不使用持久化引擎。

一、在master的OnStart方法中，对应的源码如下:

[Scala] 纯文本查看 复制代码

val serializer = new JavaSerializer(conf)
val (persistenceEngine_, leaderElectionAgent_) = RECOVERY_MODE match {
  case "ZOOKEEPER" =>
    logInfo("Persisting recovery state to ZooKeeper")
    val zkFactory =
      new ZooKeeperRecoveryModeFactory(conf, serializer)
    (zkFactory.createPersistenceEngine(), zkFactory.createLeaderElectionAgent(this))
  case "FILESYSTEM" =>
    val fsFactory =
      new FileSystemRecoveryModeFactory(conf, serializer)
    (fsFactory.createPersistenceEngine(), fsFactory.createLeaderElectionAgent(this))
  case "CUSTOM" =>
    val clazz = Utils.classForName(conf.get("spark.deploy.recoveryMode.factory"))
    val factory = clazz.getConstructor(classOf[SparkConf], classOf[Serializer])
      .newInstance(conf, serializer)
      .asInstanceOf[StandaloneRecoveryModeFactory]
    (factory.createPersistenceEngine(), factory.createLeaderElectionAgent(this))
  case _ =>
    (new BlackHolePersistenceEngine(), new MonarchyLeaderAgent(this))
}
persistenceEngine = persistenceEngine_
leaderElectionAgent = leaderElectionAgent_

      默认情况下是无持久化引擎，也就是没有ha策略。Spark提供的可用的ha策略：基于文件系统的和基于zookeeper。配置方法如下:

基于文件系统：

property	Meaning
spark.deploy.recoveryMode	FILESYSTEM
spark.deploy.recoveryDirectory	用来恢复状态的目录

基于zookeeper:

property	Meaning
spark.deploy.recoveryMode	ZOOKEEPER
spark.deploy.zookeeper.url	e.g., 192.168.1.100:2181,192.168.1.101:2181
spark.deploy.zookeeper.dir	zookeeper保存恢复状态的目录

      生产环境中可用的是基于zookeeper的持久化引擎。基于zookeeper持久化策略，会允许我们同时运行多个master，然后支持leader选举，最终是一个leader，其余是standby。

二、Spark的Master的leader选举实现

      Spark源码里面使用的是CuratorFramework，跟zookeeper交流。该框架有以下特点:

      1、自动连接管理：自动处理zookeeper的连接和重试存在一些潜在的问题；可以watch NodeDataChanged event和获取updateServerList;Watches可以自动被Cruator recipes删除；

      2、更加简洁的API：简化raw zookeeper方法，事件等，提供现代流式API接口；

      3、Recipe实现：leader选举，分布式锁，path缓存，和watcher,分布式队列，Barriers等。

三、Spark源码里面使用了LeaderLatch实现选举功能。

     这个实现实际是基于zookeeper的节点类型来做，zookeeper有四种节点类型:

     1、持久节点（PERSISTENT）

节点创建后，会一直存在，不会因客户端会话失效而删除；

     2、持久顺序节点（PERSISTENT_SEQUENTIAL）

基本特性与持久节点一致，创建节点的过程中，zookeeper会在其名字后自动追加一个单调增长的数字后缀，作为新的节点名；

     3、临时节点（EPHEMERAL）

客户端会话失效或连接关闭后，该节点会被自动删除，且不能再临时节点下面创建子节点；

     4、临时顺序节点（EPHEMERAL_SEQUENTIAL）

基本特性与临时节点一致，创建节点的过程中，zookeeper会在其名字后自动追加一个单调增长的数字后缀，作为新的节点名；LeaderLatch实现leader选举实际上基于临时顺序节点来做的。

四、Spark源码里面基于zookeeper的leader选举具体实现过程源码如下:

      在master的OnStart方法里面

[Scala] 纯文本查看 复制代码

leaderElectionAgent = leaderElectionAgent_

      实际是在构建zookeeper的持久化引擎的时候构建的:

[Scala] 纯文本查看 复制代码

(zkFactory.createPersistenceEngine(), zkFactory.createLeaderElectionAgent(this))

      在createLeaderElectionAgent方法里面构建了

[Scala] 纯文本查看 复制代码

new ZooKeeperLeaderElectionAgent(master, conf)

      该对象继承了LeaderLatchListener，并且覆盖了notLeader和isLeader两个重要的方法具体。

在ZooKeeperLeaderElectionAgent构建的时候调用了自己的start方法，该方法构建了LeaderLatch，并添加ZooKeeperLeaderElectionAgent作为其listener。

[Scala] 纯文本查看 复制代码

private def start() {
  logInfo("Starting ZooKeeper LeaderElection agent")
  zk = SparkCuratorUtil.newClient(conf)
  leaderLatch = new LeaderLatch(zk, WORKING_DIR)
  leaderLatch.addListener(this)
  leaderLatch.start()
}

      Leader选举在zookeeper的临时节点的路径为:

[Scala] 纯文本查看 复制代码

val WORKING_DIR = conf.get("spark.deploy.zookeeper.dir", "/spark") + "/leader_election"

      执行ZooKeeperLeaderElectionAgent对象的start方法之后，每当该对象所在的master由standby变为Leader的时候，会调用isLeader()方法。由Leader变为StandBy的时候会调用notLeader()。我们就可以在这两个方法里实现自己要的状态切换的相关操作。

[Scala] 纯文本查看 复制代码

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)
  }
}

要实现我们自己应用，也可基于此方法。

感谢分享