Spark PersistenceEngine持久化引擎與領導選舉代理機制內核原理深刻剖析-Spark商業環境實戰

本套系列博客從真實商業環境抽取案例進行總結和分享，並給出Spark源碼解讀及商業實戰指導，請持續關注本套博客。版權聲明：本套Spark源碼解讀及商業實戰歸做者（秦凱新）全部，禁止轉載，歡迎學習。

• Spark商業環境實戰-Spark內置框架rpc通信機制及RpcEnv基礎設施
• Spark商業環境實戰-Spark事件監聽總線流程分析
• Spark商業環境實戰-Spark存儲體系底層架構剖析
• Spark商業環境實戰-Spark底層多個MessageLoop循環線程執行流程分析
• Spark商業環境實戰-Spark一級資源調度Shedule機制及SpreadOut模式源碼深刻剖析
• Spark商業環境實戰-Spark二級調度系統Stage劃分算法和最佳任務調度細節剖析
• Spark商業環境實戰-Spark任務延遲調度及調度池Pool架構剖析
• Spark商業環境實戰-Task粒度的緩存聚合排序結構AppendOnlyMap詳細剖析
• Spark商業環境實戰-ExternalSorter 外部排序器在Spark Shuffle過程當中設計思路剖析
• Spark商業環境實戰-ShuffleExternalSorter外部排序器在Spark Shuffle過程當中的設計思路剖析
• Spark商業環境實戰-Spark ShuffleManager內存緩衝器SortShuffleWriter設計思路剖析
• Spark商業環境實戰-Spark ShuffleManager內存緩衝器UnsafeShuffleWriter設計思路剖析
• Spark商業環境實戰-Spark ShuffleManager內存緩衝器BypassMergeSortShuffleWriter設計思路剖析
• Spark商業環境實戰-Spark Shuffle 核心組件BlockStoreShuffleReader內核原理深刻剖析
• Spark商業環境實戰-Spark Shuffle 管理器SortShuffleManager內核原理深刻剖析
• Spark商業環境實戰-Spark PersistenceEngine持久化引擎高可用機制內核原理深刻剖析
• Spark商業環境實戰-StreamingContext啓動流程及Dtream 模板源碼剖析
• Spark商業環境實戰-ReceiverTracker 啓動過程及接收器 receiver RDD 任務提交機制源碼剖析
• Spark商業環境實戰-SparkStreaming數據流從Batch到Block定時轉化過程源碼深度剖析
• Spark商業環境實戰-SparkStreaming之JobGenerator週期性任務數據處理邏輯源碼深度剖析
• [Spark商業環境實戰-SparkStreaming Graph 處理鏈迭代過程源碼深度剖析]
• [Spark商業環境實戰-JobGenerator 數據清理流程源碼深度剖析]
• [Spark商業環境實戰-SparkStreaming 容錯機制源碼深度剖析]
• [Spark商業環境實戰-SparkStreaming 之No Receiver方式基於Kafka 拉取內幕源碼深度剖析]
• [Spark商業環境實戰-SparkStreaming 反壓機制控制消費速率內幕源碼深度剖析]

1 PersistenceEngine持久化引擎

1.1 PersistenceEngine的啓動

• 選擇故障恢復機制，主要有ZOOKEEPER 和 FILESYSTEM 。

  private val RECOVERY_MODE = conf.get("spark.deploy.recoveryMode", "NONE")
  複製代碼

• PersistenceEngine 的初始化是放在Master的onStart()方法中，用於初始化持久化引擎。

  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_
  複製代碼

1.2 PersistenceEngine的功能

• PersistenceEngine主要用於當Master發生故障時，來讀取持久化的Application，Worker，Driver的詳細信息。
• PersistenceEngine一樣負責寫入持久化Application，Worker，Driver的詳細信息。

（1）PersistenceEngine 的調用時機：

• 在新的Application註冊以前。
• 在新的Worker註冊以前。
• 在removeApplication和removeWorker方法被調用的時候

舉例以下：

persistenceEngine.removeWorker(worker)
複製代碼

1.3 PersistenceEngine的抽象模板，也即調用時機

abstract class PersistenceEngine {

  /**
   * Defines how the object is serialized and persisted. Implementation will
   * depend on the store used.
   */
  def persist(name: String, obj: Object): Unit

  /**
   * Defines how the object referred by its name is removed from the store.
   */
  def unpersist(name: String): Unit

  /**
   * Gives all objects, matching a prefix. This defines how objects are
   * read/deserialized back.
   */
  def read[T: ClassTag](prefix: String): Seq[T]

  final def addApplication(app: ApplicationInfo): Unit = {
    persist("app_" + app.id, app)
  }

  final def removeApplication(app: ApplicationInfo): Unit = {
    unpersist("app_" + app.id)
  }

  final def addWorker(worker: WorkerInfo): Unit = {
    persist("worker_" + worker.id, worker)
  }

  final def removeWorker(worker: WorkerInfo): Unit = {
    unpersist("worker_" + worker.id)
  }

  final def addDriver(driver: DriverInfo): Unit = {
    persist("driver_" + driver.id, driver)
  }

  final def removeDriver(driver: DriverInfo): Unit = {
    unpersist("driver_" + driver.id)
  }

  /**
   * Returns the persisted data sorted by their respective ids (which implies that they're
   * sorted by time of creation).
   */
  final def readPersistedData(
      rpcEnv: RpcEnv): (Seq[ApplicationInfo], Seq[DriverInfo], Seq[WorkerInfo]) = {
    rpcEnv.deserialize { () =>
      (read[ApplicationInfo]("app_"), read[DriverInfo]("driver_"), read[WorkerInfo]("worker_"))
    }
  }

  def close() {}
}
複製代碼

1.4 PersistenceEngine 的基於文件系統持久化和基於Zookeeper的持久化

• 基於文件系統持久化FileSystemPersistenceEngine

  private def serializeIntoFile(file: File, value: AnyRef) {
        val created = file.createNewFile()
        if (!created) { throw new IllegalStateException("Could not create file: " + file) }
        val fileOut = new FileOutputStream(file)
        var out: SerializationStream = null
        Utils.tryWithSafeFinally {
          out = serializer.newInstance().serializeStream(fileOut)
          out.writeObject(value)
        } {
          fileOut.close()
          if (out != null) {
            out.close()
          }
        }
      }
  複製代碼

• 基於Zookeeper的持久化ZooKeeperPersistenceEngine

  Curator是Netflix公司開源的Zookeeper客戶端，注意這裏會把ApplicationInfo，WorkerInfo，DriverInfo等數據經過ZooKeeperPersistenceEngine將數據存儲到Zookeeper的不一樣Znode節點上。

  這裏Zookeeper能撐得住嗎？？疑問

  private val WORKING_DIR = conf.get("spark.deploy.zookeeper.dir", "/spark") + "/master_status"
  private val zk: CuratorFramework = SparkCuratorUtil.newClient(conf)
  
    private def serializeIntoFile(path: String, value: AnyRef) {
    
        val serialized = serializer.newInstance().serialize(value)
        val bytes = new Array[Byte](serialized.remaining())
        serialized.get(bytes)
        zk.create().withMode(CreateMode.PERSISTENT).forPath(path, bytes)
     }
  複製代碼

2 領導選舉機制

所謂選舉機制就是註冊監聽機制，一旦監聽到Master掛了，就會進行回調監聽。

主要有：

• ZooKeeperLeaderElectionAgent
• MonarchyLeaderAgent

接下來主要以ZooKeeperLeaderElectionAgent爲例:

2.1 借雞生蛋的道理

• 經過/leader_election這個目錄進行監聽：

  val WORKING_DIR = conf.get("spark.deploy.zookeeper.dir", "/spark") + "/leader_election"
  
  private def start() {
      logInfo("Starting ZooKeeper LeaderElection agent")
      zk = SparkCuratorUtil.newClient(conf)
      leaderLatch = new LeaderLatch(zk, WORKING_DIR)
      leaderLatch.addListener(this)
      leaderLatch.start()
    }
  
    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()
      }
    }
  複製代碼

• 經過監聽/leader_election對應目錄來進行選舉

  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)
        }
      }
  複製代碼

3 Master 在選舉中要作什麼

Master本身給本身發送消息，開始進行恢復操做：

• Master繼承了LeaderElectable，所以實現了electedLeader方法：

  override def electedLeader() {
    self.send(ElectedLeader)
  }
  複製代碼

• Master 的行動beginRecovery和CompleteRecovery

  override def receive: PartialFunction[Any, Unit] = {
        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)
          }
  複製代碼

• Master 的行動beginRecovery

  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")
          }
        }
  複製代碼

• Master 的行動completeRecovery

  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(_, "Not responding for recovery"))
        apps.filter(_.state == ApplicationState.UNKNOWN).foreach(finishApplication)
    
        // Update the state of recovered apps to RUNNING
        apps.filter(_.state == ApplicationState.WAITING).foreach(_.state = ApplicationState.RUNNING)
    
        // 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")
          }
        }
  複製代碼

4 總結

秦凱新 於深圳