Master 与 Worker

在 Standalone 模式下 Spark 使用 Master 来进行资源的分配与管理，Worker 为具体执行任务(提供计算服务)的节点。

流程概览

Master 在 SparkCore 的 org.apache.spark.deploy.master 包下，Worker 在 org.apache.spark.deploy.worker 包下。

Master

Master 维护着 Application 和 Worker 的信息，当有新的 Application 注册过来的时候为其找到合适的 Worker 并为其在 Worker 上启动 Executor。

流程概览

Master 的 main() 方法 (执行启动 Spark 命令)：

private[deploy] object Master extends Logging {

  def main(argStrings: Array[String]) {
    Utils.initDaemon(log)
    val conf = new SparkConf
    val args = new MasterArguments(argStrings, conf)
    val (rpcEnv, _, _) = startRpcEnvAndEndpoint(args.host, args.port, args.webUiPort, conf)
    rpcEnv.awaitTermination()
  }

  def startRpcEnvAndEndpoint(
      host: String,
      port: Int,
      webUiPort: Int,
      conf: SparkConf): (RpcEnv, Int, Option[Int]) = {
   
    // RPC 环境
    val rpcEnv = RpcEnv.create(SYSTEM_NAME, host, port, conf, securityMgr)
    val masterEndpoint = rpcEnv.setupEndpoint(ENDPOINT_NAME,
      // 实例化 Master
      new Master(rpcEnv, rpcEnv.address, webUiPort, securityMgr, conf))
    val portsResponse = masterEndpoint.askSync[BoundPortsResponse](BoundPortsRequest)
    (rpcEnv, portsResponse.webUIPort, portsResponse.restPort)
  }
}

接下来，我们先看几个 Master 的成员变量，看看 Master 维护着哪些信息：

// Workers 信息
val workers = new HashSet[WorkerInfo]
private val idToWorker = new HashMap[String, WorkerInfo]
private val addressToWorker = new HashMap[RpcAddress, WorkerInfo]

// Applications 信息
val apps = new HashSet[ApplicationInfo]
val idToApp = new HashMap[String, ApplicationInfo]
private val waitingApps = new ArrayBuffer[ApplicationInfo]
private val endpointToApp = new HashMap[RpcEndpointRef, ApplicationInfo]
private val addressToApp = new HashMap[RpcAddress, ApplicationInfo]
private val completedApps = new ArrayBuffer[ApplicationInfo]

Master 的初始化工作是在 onStart() 方法中完成的：

override def onStart(): Unit = {
  // Web UI 配置
  webUi = new MasterWebUI(this, webUiPort)
  webUi.bind()
  masterWebUiUrl = "http://" + masterPublicAddress + ":" + webUi.boundPort

  // 设置定期检查 Worker 是否超时任务
  checkForWorkerTimeOutTask = forwardMessageThread.scheduleAtFixedRate(new Runnable {
    override def run(): Unit = Utils.tryLogNonFatalError {
      self.send(CheckForWorkerTimeOut)
    }
  }, 0, WORKER_TIMEOUT_MS, TimeUnit.MILLISECONDS)

  // 其它操作

  // 启动指标系统
  masterMetricsSystem.registerSource(masterSource)
  masterMetricsSystem.start()
  applicationMetricsSystem.start()
  // 指标系统绑定到 Web UI
  masterMetricsSystem.getServletHandlers.foreach(webUi.attachHandler)
  applicationMetricsSystem.getServletHandlers.foreach(webUi.attachHandler)

  // 根据模式的不同，选择不同的持久化机制
  val (persistenceEngine_, leaderElectionAgent_) = RECOVERY_MODE match {
      
    // 使用 Zookeeper
    case "ZOOKEEPER" =>
      val zkFactory =
        new ZooKeeperRecoveryModeFactory(conf, serializer)
      (zkFactory.createPersistenceEngine(), zkFactory.createLeaderElectionAgent(this))

    // 其它的模式
      
  }
  persistenceEngine = persistenceEngine_
  // leader 选举
  leaderElectionAgent = leaderElectionAgent_
}

Master 在初始化的时候主要做了两件事，一是启动 Web UI，二是持久化和选举模式的选择。

当 Worker 启动的时候，会向 Master 发送一条注册信息，然后 Master 将其加入到 Worker 集合中(成员变量 workers)，响应 Worker 注册请求的代码可以在 Master.receive() 方法中找到：

override def receive: PartialFunction[Any, Unit] = {
  // 选举机制
  case ElectedLeader =>
    // 略略略

  // Master 主备切换
  // 有兴趣的可以看看
  case CompleteRecovery => completeRecovery()

  // Worker 注册
  case RegisterWorker(
    id, workerHost, workerPort, workerRef, cores, memory, workerWebUiUrl, masterAddress) =>
    
    if (state == RecoveryState.STANDBY) {
      workerRef.send(MasterInStandby)
    } else if (idToWorker.contains(id)) {
      workerRef.send(RegisterWorkerFailed("Duplicate worker ID"))
    } else {
      // 主要看这里
      // 将 Worker 发送过来的信息进行了封装
      val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,
        workerRef, workerWebUiUrl)
      // 注册 Worker
      if (registerWorker(worker)) {
        // 持久化 Worker 信息
        // 比如使用 Zookeepeer
        persistenceEngine.addWorker(worker)
        // 向 Worker 发送一条确认信息
        workerRef.send(RegisteredWorker(self, masterWebUiUrl, masterAddress))
        // 编制执行计划
        // 因为有新 Worker 的到达，所以可以唤醒一些等待执行的 Application
        // 等会看内部源码
        schedule()
      } else {
          // 略略略...
      }
    }

  // Application 注册
  case RegisterApplication(description, driver) =>
    // 等会看

  // Executor 状态改变
  case ExecutorStateChanged(appId, execId, state, message, exitStatus) =>
    // 略略略...

  // Driver 状态改变
  case DriverStateChanged(driverId, state, exception) =>
    // 略略略...

  // 心跳机制
  case Heartbeat(workerId, worker) =>
    // 略略略...

  // Worker 超时检查
  case CheckForWorkerTimeOut =>
    timeOutDeadWorkers()

}

Master.registerWorker() 的实现细节：

private def registerWorker(worker: WorkerInfo): Boolean = {

  // Worker 集合中可能已经注册过这个 Worker (发生了故障)
  // 将其移出
  workers.filter { w =>
    (w.host == worker.host && w.port == worker.port) && (w.state == WorkerState.DEAD)
  }.foreach { w =>
    workers -= w
  }

  // Worker RPC Address
  // 补充 addressToWorker 信息 (成员变量)
  val workerAddress = worker.endpoint.address
  if (addressToWorker.contains(workerAddress)) {
    // 略略略
  }
    
  // 将 Worker 放入到 Worker 集合
  workers += worker
  
  // 补充 idToWorker 信息 (成员变量)
  idToWorker(worker.id) = worker
  addressToWorker(workerAddress) = worker
  if (reverseProxy) {
     // Web UI 添加
     webUi.addProxyTargets(worker.id, worker.webUiAddress)
  }
  true
}

在收到 Worker 发送过来的注册信息后，Master 会将 Worker 的信息添加到自身的成员变量中，并且还会将 Worker 的信息持久化一份 (当 Master 挂了之后可以恢复)。

接下来我们看看，在 Master 收到 Application 发送过来的注册消息后会做哪些工作，同样可以在 Master.receive() 方法中可以找到：

override def receive: PartialFunction[Any, Unit] = {

  // Application 注册
  case RegisterApplication(description, driver) =>

    if (state == RecoveryState.STANDBY) {
      // ignore, don't send response
    } else {
      // 构建 Application
      val app = createApplication(description, driver)
      // 注册 App
      // 主要是添加到等待队列，并记录 App 信息，上文提到的成员变量
      // 与注册 Worker 类似
      registerApplication(app)
      // 持久化
      persistenceEngine.addApplication(app)
      // 向 Driver 发送一条确认信息
      driver.send(RegisteredApplication(app.id, self))
      // 有新的 App 的加入
      // 所以需要重新编排下执行计划
      // 如果等待执行队列为空，那么这个 App 可能就直接被 Master 调度
      schedule()
    }

}

在注册完 Worker 和 Application 后，最后都调用了 Master.schedule() 方法来编排执行计划，接下来我们看看其实现细节：

private def schedule(): Unit = {
    
  // 对活跃的 Worker 进行随机打散
  // 尽量负载均衡
  val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(_.state == WorkerState.ALIVE))
  // 活跃的 Worker 数
  val numWorkersAlive = shuffledAliveWorkers.size
  var curPos = 0
    
  // 针对 Driver 的集群启动模式
  for (driver <- waitingDrivers.toList) {
    var launched = false
    var numWorkersVisited = 0
    // 循环遍历活跃的 Worker，直到某个 Worker 满足 Driver 的资源申请
    while (numWorkersVisited < numWorkersAlive && !launched) {
      val worker = shuffledAliveWorkers(curPos)
      numWorkersVisited += 1
      // 当前的 Worker 可以满足 Driver 申请的资源数
      if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
        // 启动 Driver
        launchDriver(worker, driver)
        waitingDrivers -= driver
        launched = true
      }
      curPos = (curPos + 1) % numWorkersAlive
    }
  }
    
  // Application 的 Executor 的分配机制
  startExecutorsOnWorkers()
}

我这里关注的重点是 Master 如何对 Application 的进行资源分配的，所以我们继续看下 startExecutorsOnWorkers() 的实现细节：

private def startExecutorsOnWorkers(): Unit = {

  // Master 使用先进先出的调度方式
  for (app <- waitingApps if app.coresLeft > 0) {
    // App 需要每个 Executor CPU 的核数
    // 在提交任务的时候由我们指定的
    val coresPerExecutor: Option[Int] = app.desc.coresPerExecutor
    // 在活跃的 Workers 中找出满足 App 资源需求的 Worker
    val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE)
      .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB &&
        worker.coresFree >= coresPerExecutor.getOrElse(1))
      .sortBy(_.coresFree).reverse
      
    // scheduleExecutorsOnWorkers() 这个方法有点复杂
    // 计算在哪些 Worker 上启动 Executor
    // 返回可执行的 Worker 数组及内核数
    // 内部有两种分配方式：
    //   1. 多启动 Worker，让 Worker 均摊提供给 Executor 的 Core.
    //   2. 少启动 Worker，让 Worker 提供最大的 Core 给 Executor（最大的 Core 为
    //      每个 Executor 需要的Core）.
    // 有兴趣的可以自己看下
    val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps)

    // 对 Worker 进行资源分配
    for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) {
      allocateWorkerResourceToExecutors(
        app, assignedCores(pos), coresPerExecutor, usableWorkers(pos))
    }
  }
}

我们还是看点简单的，看看 allocateWorkerResourceToExecutors() 的内部实现细节：

private def allocateWorkerResourceToExecutors(
    app: ApplicationInfo,
    assignedCores: Int,
    coresPerExecutor: Option[Int],
    worker: WorkerInfo): Unit = {
  
  // 在这个 Worker 上需要启动的 Executor 数
  val numExecutors = coresPerExecutor.map { assignedCores / _ }.getOrElse(1)
  // 可分配的 Core 数量
  val coresToAssign = coresPerExecutor.getOrElse(assignedCores)
  for (i <- 1 to numExecutors) {
    // Executor 信息
    val exec = app.addExecutor(worker, coresToAssign)
    // 在 Worker 上启动 Executor
    launchExecutor(worker, exec)
    app.state = ApplicationState.RUNNING
  }
}

launchExecutor() 的实现细节:

private def launchExecutor(worker: WorkerInfo, exec: ExecutorDesc): Unit = {
  worker.addExecutor(exec)
  // 向 Worker 发送启动 Executor 的消息
  worker.endpoint.send(LaunchExecutor(masterUrl,
    exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory))
  // 向 Driver 发送分配完成消息
  // Executor 在启动完成后会主动联系 Driver
  exec.application.driver.send(
    ExecutorAdded(exec.id, worker.id, worker.hostPort, exec.cores, exec.memory))
}

简单的总结一下，Master 会维护着 Application 和 Worker 的信息，当有新的 Application 过来注册时，Master 会从 Worker 集合中找到合适的 Worker 并为其分配并启动 Executor。

Worker

Worker 是具体提供计算节点，当收到 Master 的启动 Executor 启动后，就会启动 Executor 进程。

流程概览

我们还是先看看 Worker 维护着哪些信息：

private val workerId = generateWorkerId()
val drivers = new HashMap[String, DriverRunner]
val executors = new HashMap[String, ExecutorRunner]
val finishedDrivers = new LinkedHashMap[String, DriverRunner]
val appDirectories = new HashMap[String, Seq[String]]
val finishedApps = new HashSet[String]

接下来我们从 Worker 的 onStart() 方法入手，看看 Worker 做了哪些工作：

override def onStart() {
  // 创建工作目录
  createWorkDir()
  // Web UI
  shuffleService.startIfEnabled()
  webUi = new WorkerWebUI(this, workDir, webUiPort)
  webUi.bind()

  workerWebUiUrl = s"http://$publicAddress:${webUi.boundPort}"
    
  // 向 Master 进行注册
  // 内部实现会调用 tryRegisterAllMasters()，与 SchedulerBackend 有点像
  registerWithMaster()

  // 指标信息绑定界面
  metricsSystem.registerSource(workerSource)
  metricsSystem.start()
  metricsSystem.getServletHandlers.foreach(webUi.attachHandler)
}

接下来我们看看，当 Worker 收到 Master 的启动 Executor 请求后会做哪些工作，可以在 Worker.receive() 方法中找到：

override def receive: PartialFunction[Any, Unit] = synchronized {
  case msg: RegisterWorkerResponse =>
    handleRegisterResponse(msg)

  // 发送心跳
  case SendHeartbeat =>
    if (connected) { sendToMaster(Heartbeat(workerId, self)) }

  // 其它的都忽略

  // 启动 Executor
  case LaunchExecutor(masterUrl, appId, execId, appDesc, cores_, memory_) =>
    if (masterUrl != activeMasterUrl) {
      logWarning("Invalid Master (" + masterUrl + ") attempted to launch executor.")
    } else {
      try {
        // 创建 Executor 工作目录
        val executorDir = new File(workDir, appId + "/" + execId)
        if (!executorDir.mkdirs()) {
          throw new IOException("Failed to create directory " + executorDir)
        }

        // 创建本地工作目录
        val appLocalDirs = appDirectories.getOrElse(...)
        
        // 创建一个 ExecutorRunner 负责管理
        val manager = new ExecutorRunner(...)
        
        // 启动 Executor
        // 内部使用 ProcessBuilder 去启动 Executor (通过启动命令)
        manager.start()
        
        // 更新自身的资源使用情况
        coresUsed += cores_
        memoryUsed += memory_
        
        // 给 Master 一条确认信息
        sendToMaster(ExecutorStateChanged(appId, execId, manager.state, None, None))
      } catch {
        // 略略略...
      }
    }


  // 这个是针对集群模式 Driver 的
  case LaunchDriver(driverId, driverDesc) =>
 
    val driver = new DriverRunner(...)
    // 启动 Driver
    // 与 Executor 类似，内部使用 ProcessBuilder 去启动
    driver.start()

    coresUsed += driverDesc.cores
    memoryUsed += driverDesc.mem

}

ExecutorRunner.start() 的实现细节：

private[worker] def start() {
  // 启动一个线程去处理
  workerThread = new Thread("ExecutorRunner for " + fullId) {
    override def run() { fetchAndRunExecutor() }
  }
  workerThread.start()

  // 略略略
}

fetchAndRunExecutor() 的实现细节：

// 下载并运行 Executor
private def fetchAndRunExecutor() {
  try {
    // 启动一个进程
    val builder = CommandUtils.buildProcessBuilder(appDesc.command, new SecurityManager(conf),
      memory, sparkHome.getAbsolutePath, substituteVariables)
    
    // 略略略
    
    // 启动进程
    process = builder.start()

    // 略略略
      
  } catch {
    // 略略略
  }
}

ProcessBuilder.start() 主要就是通过命令启动一个新的进程，内部实现太复杂了，这里就不看了。

简单的总结一下，我们只看了很少的 Worker 源码，也就是接收到 Master 的启动 Executor 请求，然后启动 Executor (新进程)。