SchedulerBackend 源码浅析

SchedulerBackend

SparkContext 使用 SchedulerBackend 来与不同资源管理系统对接，并维护着 Executor 的信息，TaskScheduler 向 Executor 提交任务也是通过 SchedulerBackend 来完成的。

SchedulerBackend 流程概览

类关系

我这里以 Standalone 模式为例，StandaloneSchedulerBackend 在 SparkCore 的 org.apache.spark.scheduler.cluster 包下。

源码

在 SaprkContext 初始化完 SchedulerBackend 和 TaskScheduler 后，调用了 TaskScheduler.start() 方法，在 SparkContext 概览中提到过。

_taskScheduler.start()

在这个版本中 TaskScheduler 只有一个实现类 TaskSchedulerImpl，在 TaskSchedulerImpl.start() 方法中对 SchedulerBackend 进行了启动：

override def start() {
  // 这里会先调用 SchedulerBackend.start() 方法
  backend.start()

  // 其它代码忽略
}

这样，我们就找到了 SchedulerBackend 的入口。

在 StandaloneSchedulerBackend. start() 方法中，我们只关注两个操作，一是调用父类的 start() 方法，二是初始化并启动 AppClient：

override def start() {
  
  // 注意这里
  super.start()
    
  // ...

  // 应用的信息
  val appDesc = ApplicationDescription(sc.appName, maxCores, sc.executorMemory, command,
    webUrl, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor, initialExecutorLimit)
  // 注意这里
  // 创建了一个 Application 客户端  
  client = new StandaloneAppClient(sc.env.rpcEnv, masters, appDesc, this, conf)
  client.start()
    
  // ..
    
}

按照先进先看原则，我们先看 super.start() 内部实现细节，也就是 CoarseGrainedSchedulerBackend.start() 方法：

override def start() {
  // Driver 通信端点
  driverEndpoint = createDriverEndpointRef(properties)
}

createDriverEndpointRef() 的实现细节：

protected def createDriverEndpointRef(
    properties: ArrayBuffer[(String, String)]): RpcEndpointRef = {

  // 同样，也将 Driver 通信端(监听器)注册到 RpcEnv 中
  // 监听器模式
  rpcEnv.setupEndpoint(ENDPOINT_NAME, createDriverEndpoint(properties))
    
}

createDriverEndpoint() 的实现细节：

protected def createDriverEndpoint(properties: Seq[(String, String)]): DriverEndpoint = {
  new DriverEndpoint(rpcEnv, properties)
}

向 RpcEnv 中注册的是 DriverEndpoint 对象，我们再看看 DriverEndpoint 中的一个成员变量：

// 维护着 Executor 的信息
private val executorDataMap = new HashMap[String, ExecutorData]

接下来，我们看看 DriverEndpoint 会对哪些消息进行处理，匹配消息的代码在 DriverEndpoint.receive() 和 DriverEndpoint.receiveAndReply() 方法中可以找到：

// 这里只分析 Executor 的反向注册，别的都忽略

override def receive: PartialFunction[Any, Unit] = {
  case StatusUpdate(executorId, taskId, state, data) =>
    // ...

  // 这个比较重要，后面会说到
  case ReviveOffers =>
    makeOffers()

  case KillTask(taskId, executorId, interruptThread, reason) =>
    // ...

  case KillExecutorsOnHost(host) =>
    //...
}

// 处理需要应答的请求
override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
    
  // 处理 Executor 的反向注册
  case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls) =>
    if (executorDataMap.contains(executorId)) {
        // ...
    } else if (scheduler.nodeBlacklist != null &&
        // ...
    } else {
      // Executor 信息
      val data = new ExecutorData(executorRef, executorAddress, hostname,
        cores, cores, logUrls)
        
      // 将 Executor 信息加入到 Executor 字典中
      // 这里使用 synchronized 是因为在添加的时候，可能会有人在读取...
      CoarseGrainedSchedulerBackend.this.synchronized {
        executorDataMap.put(executorId, data)
        // ...
      }
      // 给 Executor 返回一条信息
      // 后面的文章在看 Executor 源码的时候会看到
      executorRef.send(RegisteredExecutor)
      
      // ...
    }
               
  case StopDriver =>
    // ...

  case StopExecutors =>
    // ...

  case RemoveExecutor(executorId, reason) =>
    // ...

  case RetrieveSparkAppConfig =>
    // ... 

}

super.start() 相关代码就看到这里，接下来，我们看下 client.start() 的实现细节，也就是 StandaloneAppClient.start() 的实现细节：

def start() {
  // 又是向 RpcEnv 中注册 Client 通信端
  endpoint.set(rpcEnv.setupEndpoint("AppClient", new ClientEndpoint(rpcEnv)))
}

我们先看下 ClientEndpoint.onStart() 方法：

override def onStart(): Unit = {
  try {
     // 向 Master 进行注册
    registerWithMaster(1)
  } catch {
    // 其他操作
  }
}

registerWithMaster() 会向 Master 发送一条 RegisterApplication 消息，来进行注册：

// registerWithMaster() 会先调用 tryRegisterAllMasters() 方法
// 这行语句在 tryRegisterAllMasters() 方法中
// 向 Master 注册应用
masterRef.send(RegisterApplication(appDescription, self))

当 Master 收到注册申请后，会返回一条 RegisteredApplication 消息（剖析 Master 的时候会看到）。

接下来，我们就看看 ClientEndpoint.receive() 和 ClientEndpoint.receiveAndReply() 方法会对哪些消息进行处理：

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

  // Master 的响应消息
  case RegisteredApplication(appId_, masterRef) =>
    appId.set(appId_)
    registered.set(true)
    // 设置 Master
    master = Some(masterRef)
    // 监听
    listener.connected(appId.get)

  case ApplicationRemoved(message) =>
    // ...

  case ExecutorAdded(id: Int, workerId: String, hostPort: String, cores: Int, memory: Int) =>
    // ...

  case ExecutorUpdated(id, state, message, exitStatus, workerLost) =>
    // ...

  case MasterChanged(masterRef, masterWebUiUrl) =>
    // ...
}

override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
  
  case StopAppClient =>
    // ...

  case r: RequestExecutors =>
    // ...

  case k: KillExecutors =>
    // ...
}

简单的总结下，SchedulerBackend 在启动的时候会创建 DriverEndpoint 和 ClientEndpoint。DriverEndpoint 维护着注册过来的 Executor 信息，这也正是 TaskScheduler 提交任务所需要的；ClientEndpoint 会向 Master 发送一条注册应用的消息，这样 Master 才会为这个程序分配 Worker 去启动 Executor。