spark通信流程

概述

spark作為一套高效的分布式運(yùn)算框架，但是想要更深入的學(xué)習(xí)它，就要通過分析spark的源碼，不但可以更好的幫助理解spark的工作過程，還可以提高對(duì)集群的排錯(cuò)能力，本文主要關(guān)注的是Spark的Master的啟動(dòng)流程與Worker啟動(dòng)流程。

現(xiàn)在Spark最新版本為1.6，但是代碼的邏輯不夠清晰，不便于理解，這里以1.3為準(zhǔn)

Master啟動(dòng)

我們啟動(dòng)一個(gè)Master是通過Shell命令啟動(dòng)了一個(gè)腳本start-master.sh開始的，這個(gè)腳本的啟動(dòng)流程如下

start-master.sh  -> spark-daemon.sh start org.apache.spark.deploy.master.Master

我們可以看到腳本首先啟動(dòng)了一個(gè)org.apache.spark.deploy.master.Master類，啟動(dòng)時(shí)會(huì)傳入一些參數(shù)，比如cpu的執(zhí)行核數(shù)，內(nèi)存大小，app的main方法等

查看Master類的main方法

private[spark] object Master extends Logging {
  val systemName = "sparkMaster"
  private val actorName = "Master"

  //master啟動(dòng)的入口
  def main(argStrings: Array[String]) {
    SignalLogger.register(log)
    //創(chuàng)建SparkConf
    val conf = new SparkConf
    //保存參數(shù)到SparkConf
    val args = new MasterArguments(argStrings, conf)
    //創(chuàng)建ActorSystem和Actor
    val (actorSystem, _, _, _) = startSystemAndActor(args.host, args.port, args.webUiPort, conf)
    //等待結(jié)束
    actorSystem.awaitTermination()
  }

這里主要看startSystemAndActor方法

  /**
   * Start the Master and return a four tuple of:
   *   (1) The Master actor system
   *   (2) The bound port
   *   (3) The web UI bound port
   *   (4) The REST server bound port, if any
   */
  def startSystemAndActor(
      host: String,
      port: Int,
      webUiPort: Int,
      conf: SparkConf): (ActorSystem, Int, Int, Option[Int]) = {
    val securityMgr = new SecurityManager(conf)

    //利用AkkaUtils創(chuàng)建ActorSystem
    val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port, conf = conf,
      securityManager = securityMgr)
   
    val actor = actorSystem.actorOf(
      Props(classOf[Master], host, boundPort, webUiPort, securityMgr, conf), actorName)
   ....
  }
}

spark底層通信使用的是Akka
通過ActorSystem創(chuàng)建Actor -> actorSystem.actorOf, 就會(huì)執(zhí)行Master的構(gòu)造方法->然后執(zhí)行Actor生命周期方法
執(zhí)行Master的構(gòu)造方法初始化一些變量

 private[spark] class Master(
    host: String,
    port: Int,
    webUiPort: Int,
    val securityMgr: SecurityManager,
    val conf: SparkConf)
  extends Actor with ActorLogReceive with Logging with LeaderElectable {
  //主構(gòu)造器

  //啟用定期器功能
  import context.dispatcher   // to use Akka's scheduler.schedule()

  val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)

  def createDateFormat = new SimpleDateFormat("yyyyMMddHHmmss")  // For application IDs
  //woker超時(shí)時(shí)間
  val WORKER_TIMEOUT = conf.getLong("spark.worker.timeout", 60) * 1000
  val RETAINED_APPLICATIONS = conf.getInt("spark.deploy.retainedApplications", 200)
  val RETAINED_DRIVERS = conf.getInt("spark.deploy.retainedDrivers", 200)
  val REAPER_ITERATIONS = conf.getInt("spark.dead.worker.persistence", 15)
  val RECOVERY_MODE = conf.get("spark.deploy.recoveryMode", "NONE")

  //一個(gè)HashSet用于保存WorkerInfo
  val workers = new HashSet[WorkerInfo]
  //一個(gè)HashMap用保存workid -> WorkerInfo
  val idToWorker = new HashMap[String, WorkerInfo]
  val addressToWorker = new HashMap[Address, WorkerInfo]

  //一個(gè)HashSet用于保存客戶端（SparkSubmit）提交的任務(wù)
  val apps = new HashSet[ApplicationInfo]
  //一個(gè)HashMap Appid-》 ApplicationInfo
  val idToApp = new HashMap[String, ApplicationInfo]
  val actorToApp = new HashMap[ActorRef, ApplicationInfo]
  val addressToApp = new HashMap[Address, ApplicationInfo]
  //等待調(diào)度的App
  val waitingApps = new ArrayBuffer[ApplicationInfo]
  val completedApps = new ArrayBuffer[ApplicationInfo]
  var nextAppNumber = 0
  val appIdToUI = new HashMap[String, SparkUI]

  //保存DriverInfo
  val drivers = new HashSet[DriverInfo]
  val completedDrivers = new ArrayBuffer[DriverInfo]
  val waitingDrivers = new ArrayBuffer[DriverInfo] // Drivers currently spooled for scheduling

主構(gòu)造器執(zhí)行完就會(huì)執(zhí)行preStart --》執(zhí)行完receive方法

  //啟動(dòng)定時(shí)器，進(jìn)行定時(shí)檢查超時(shí)的worker
  //重點(diǎn)看一下CheckForWorkerTimeOut
  context.system.scheduler.schedule(0 millis, WORKER_TIMEOUT millis, self, CheckForWorkerTimeOut)

preStart方法里創(chuàng)建了一個(gè)定時(shí)器，定時(shí)檢查Woker的超時(shí)時(shí)間 val WORKER_TIMEOUT = conf.getLong("spark.worker.timeout", 60) * 1000 默認(rèn)為60秒

到此Master的初始化的主要過程到我們已經(jīng)看到了，主要就是構(gòu)造一個(gè)Master的Actor進(jìn)行等待消息，并初始化了一堆集合來保存Worker信息，和一個(gè)定時(shí)器來檢查Worker的超時(shí)

Master啟動(dòng)時(shí)序圖

Woker的啟動(dòng)

通過Shell腳本執(zhí)行salves.sh -> 通過讀取slaves 通過ssh的方式啟動(dòng)遠(yuǎn)端的worker
spark-daemon.sh start org.apache.spark.deploy.worker.Worker

腳本會(huì)啟動(dòng)org.apache.spark.deploy.worker.Worker類

看Worker源碼

private[spark] object Worker extends Logging {
  //Worker啟動(dòng)的入口
  def main(argStrings: Array[String]) {
    SignalLogger.register(log)
    val conf = new SparkConf
    val args = new WorkerArguments(argStrings, conf)
    //新創(chuàng)ActorSystem和Actor
    val (actorSystem, _) = startSystemAndActor(args.host, args.port, args.webUiPort, args.cores,
      args.memory, args.masters, args.workDir)
    actorSystem.awaitTermination()
  }

這里最重要的是Woker的startSystemAndActor

  def startSystemAndActor(
      host: String,
      port: Int,
      webUiPort: Int,
      cores: Int,
      memory: Int,
      masterUrls: Array[String],
      workDir: String,
      workerNumber: Option[Int] = None,
      conf: SparkConf = new SparkConf): (ActorSystem, Int) = {

    // The LocalSparkCluster runs multiple local sparkWorkerX actor systems
    val systemName = "sparkWorker" + workerNumber.map(_.toString).getOrElse("")
    val actorName = "Worker"
    val securityMgr = new SecurityManager(conf)
    //通過AkkaUtils ActorSystem
    val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port,
      conf = conf, securityManager = securityMgr)
    val masterAkkaUrls = masterUrls.map(Master.toAkkaUrl(_, AkkaUtils.protocol(actorSystem)))
    //通過actorSystem.actorOf創(chuàng)建Actor   Worker-》執(zhí)行構(gòu)造器 -》 preStart -》 receice
    actorSystem.actorOf(Props(classOf[Worker], host, boundPort, webUiPort, cores, memory,
      masterAkkaUrls, systemName, actorName,  workDir, conf, securityMgr), name = actorName)
    (actorSystem, boundPort)
  }

這里Worker同樣的構(gòu)造了一個(gè)屬于Worker的Actor對(duì)象,到此Worker的啟動(dòng)初始化完成

Worker與Master通信

根據(jù)Actor生命周期接著Worker的preStart方法被調(diào)用

  override def preStart() {
    assert(!registered)
    logInfo("Starting Spark worker %s:%d with %d cores, %s RAM".format(
      host, port, cores, Utils.megabytesToString(memory)))
    logInfo(s"Running Spark version ${org.apache.spark.SPARK_VERSION}")
    logInfo("Spark home: " + sparkHome)
    createWorkDir()
    context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
    shuffleService.startIfEnabled()
    webUi = new WorkerWebUI(this, workDir, webUiPort)
    webUi.bind()

    //Worker向Master注冊(cè)
    registerWithMaster()
    ....
  }

這里調(diào)用了一個(gè)registerWithMaster方法，開始向Master注冊(cè)

 def registerWithMaster() {
    // DisassociatedEvent may be triggered multiple times, so don't attempt registration
    // if there are outstanding registration attempts scheduled.
    registrationRetryTimer match {
      case None =>
        registered = false
        //開始注冊(cè)
        tryRegisterAllMasters()
        ....
    }
  }

registerWithMaster里通過匹配調(diào)用了tryRegisterAllMasters方法
，接下來看

  private def tryRegisterAllMasters() {
    //遍歷master的地址
    for (masterAkkaUrl <- masterAkkaUrls) {
      logInfo("Connecting to master " + masterAkkaUrl + "...")
      //Worker跟Mater建立連接
      val actor = context.actorSelection(masterAkkaUrl)
      //向Master發(fā)送注冊(cè)信息
      actor ! RegisterWorker(workerId, host, port, cores, memory, webUi.boundPort, publicAddress)
    }
  }

通過masterAkkaUrl和Master建立連接后
actor ! RegisterWorker(workerId, host, port, cores, memory, webUi.boundPort, publicAddress)Worker向Master發(fā)送了一個(gè)消息，帶去一些參數(shù)，id,主機(jī)，端口，cpu核數(shù)，內(nèi)存等待

override def receiveWithLogging = {
    ......
  
    //接受來自Worker的注冊(cè)信息
    case RegisterWorker(id, workerHost, workerPort, cores, memory, workerUiPort, publicAddress) =>
    {
      logInfo("Registering worker %s:%d with %d cores, %s RAM".format(
        workerHost, workerPort, cores, Utils.megabytesToString(memory)))
      if (state == RecoveryState.STANDBY) {
        // ignore, don't send response
        //判斷這個(gè)worker是否已經(jīng)注冊(cè)過
      } else if (idToWorker.contains(id)) {
        //如果注冊(cè)過，告訴worker注冊(cè)失敗
        sender ! RegisterWorkerFailed("Duplicate worker ID")
      } else {
        //沒有注冊(cè)過，把來自Worker的注冊(cè)信息封裝到WorkerInfo當(dāng)中
        val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,
          sender, workerUiPort, publicAddress)
        if (registerWorker(worker)) {
          //用持久化引擎記錄Worker的信息
          persistenceEngine.addWorker(worker)
          //向Worker反饋信息，告訴Worker注冊(cè)成功
          sender ! RegisteredWorker(masterUrl, masterWebUiUrl)
    
          schedule()
        } else {
          val workerAddress = worker.actor.path.address
          logWarning("Worker registration failed. Attempted to re-register worker at same " +
            "address: " + workerAddress)
          sender ! RegisterWorkerFailed("Attempted to re-register worker at same address: "
            + workerAddress)
        }
      }
    }

這里是最主要的內(nèi)容；
receiveWithLogging里會(huì)輪詢到Worker發(fā)送的消息，
Master收到消息后將參數(shù)封裝成WorkInfo對(duì)象添加到集合中，并加入到持久化引擎中
sender ! RegisteredWorker(masterUrl, masterWebUiUrl)向Worker發(fā)送一個(gè)消息反饋

接下來看Worker的receiveWithLogging

override def receiveWithLogging = {
    
    case RegisteredWorker(masterUrl, masterWebUiUrl) =>
      logInfo("Successfully registered with master " + masterUrl)
      registered = true
      changeMaster(masterUrl, masterWebUiUrl)
      //啟動(dòng)定時(shí)器，定時(shí)發(fā)送心跳Heartbeat
      context.system.scheduler.schedule(0 millis, HEARTBEAT_MILLIS millis, self, SendHeartbeat)
      if (CLEANUP_ENABLED) {
        logInfo(s"Worker cleanup enabled; old application directories will be deleted in: $workDir")
        context.system.scheduler.schedule(CLEANUP_INTERVAL_MILLIS millis,
          CLEANUP_INTERVAL_MILLIS millis, self, WorkDirCleanup)
      }

worker接受來自Master的注冊(cè)成功的反饋信息,啟動(dòng)定時(shí)器，定時(shí)發(fā)送心跳Heartbeat

    case SendHeartbeat =>
      //worker發(fā)送心跳的目的就是為了報(bào)活
      if (connected) { master ! Heartbeat(workerId) }

Master端的receiveWithLogging收到心跳消息

  override def receiveWithLogging = {
        ....
    case Heartbeat(workerId) => {
      idToWorker.get(workerId) match {
        case Some(workerInfo) =>
          //更新最后一次心跳時(shí)間
          workerInfo.lastHeartbeat = System.currentTimeMillis()
          .....
      }
    }
 }

記錄并更新workerInfo.lastHeartbeat = System.currentTimeMillis()最后一次心跳時(shí)間

Master的定時(shí)任務(wù)會(huì)不斷的發(fā)送一個(gè)CheckForWorkerTimeOut內(nèi)部消息不斷的輪詢集合里的Worker信息，如果超過60秒就將Worker信息移除

  //檢查超時(shí)的Worker
    case CheckForWorkerTimeOut => {
      timeOutDeadWorkers()
    }

timeOutDeadWorkers方法

  def timeOutDeadWorkers() {
    // Copy the workers into an array so we don't modify the hashset while iterating through it
    val currentTime = System.currentTimeMillis()
    val toRemove = workers.filter(_.lastHeartbeat < currentTime - WORKER_TIMEOUT).toArray
    for (worker <- toRemove) {
      if (worker.state != WorkerState.DEAD) {
        logWarning("Removing %s because we got no heartbeat in %d seconds".format(
          worker.id, WORKER_TIMEOUT/1000))
        removeWorker(worker)
      } else {
        if (worker.lastHeartbeat < currentTime - ((REAPER_ITERATIONS + 1) * WORKER_TIMEOUT)) {
          workers -= worker // we've seen this DEAD worker in the UI, etc. for long enough; cull it
        }
      }
    }
  }

如果 （最后一次心跳時(shí)間<當(dāng)前時(shí)間-超時(shí)時(shí)間）則判斷為Worker超時(shí)，
將集合里的信息移除。
當(dāng)下一次收到心跳信息時(shí)，如果是已注冊(cè)過的，workerId不為空，但是WorkerInfo已被移除的條件，就會(huì)sender ! ReconnectWorker(masterUrl)發(fā)送一個(gè)重新注冊(cè)的消息

 case None =>
          if (workers.map(_.id).contains(workerId)) {
            logWarning(s"Got heartbeat from unregistered worker $workerId." +
              " Asking it to re-register.")
            //發(fā)送重新注冊(cè)的消息
            sender ! ReconnectWorker(masterUrl)
          } else {
            logWarning(s"Got heartbeat from unregistered worker $workerId." +
              " This worker was never registered, so ignoring the heartbeat.")
          }

Worker與Master時(shí)序圖