Spark源码分析-Spark-on-K8S任务调度

Controls the propagation of the Spark application's executor pods state to subscribers that
  react to that state.
  
  Roughly follows a producer-consumer model. Producers report states of executor pods, and these
  states are then published to consumers that can perform any actions in response to these states.
  
  Producers push updates in one of two ways. An incremental update sent by updatePod() represents
  a known new state of a single executor pod. A full sync sent by replaceSnapshot() indicates that
  the passed pods are all of the most up to date states of all executor pods for the application.
  The combination of the states of all executor pods for the application is collectively known as
  a snapshot. The store keeps track of the most up to date snapshot, and applies updates to that
  most recent snapshot - either by incrementally updating the snapshot with a single new pod state,
  or by replacing the snapshot entirely on a full sync.
  
  Consumers, or subscribers, register that they want to be informed about all snapshots of the
  executor pods. Every time the store replaces its most up to date snapshot from either an
  incremental update or a full sync, the most recent snapshot after the update is posted to the
  subscriber's buffer. Subscribers receive blocks of snapshots produced by the producers in
  time-windowed chunks. Each subscriber can choose to receive their snapshot chunks at different
  time intervals.
  
  The subscriber notification callback is guaranteed to be called from a single thread at a time.

#注册监听pod事件
watchConnection = kubernetesClient.pods()
.withLabel(SPARK\\\\_APP\\\\_ID\\\\_LABEL, applicationId)
.withLabel(SPARK\\\\_ROLE\\\\_LABEL, SPARK\\\\_POD\\\\_EXECUTOR\\\\_ROLE)
.watch(new ExecutorPodsWatcher())

#事件处理：调用snapshotsStore更新
override def eventReceived(action: Action, pod: Pod): Unit = {
val podName = pod.getMetadata.getName
logDebug(s"Received executor pod update for pod named $podName, action $action")
snapshotsStore.updatePod(pod)
}

  def start(applicationId: String): Unit = {
     pollingFuture = pollingExecutor.scheduleWithFixedDelay(
     new PollRunnable(applicationId), pollingInterval, pollingInterval, TimeUnit.MILLISECONDS)
  }

 #runnable逻辑
 private class PollRunnable(applicationId: String) extends Runnable {

  override def run(): Unit = Utils.tryLogNonFatalError {

  logDebug(s"Resynchronizing full executor pod state from Kubernetes.")
  snapshotsStore.replaceSnapshot(kubernetesClient
    .pods()
    .withLabel(SPARK_APP_ID_LABEL, applicationId)
    .withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
    .withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")
    .list()
    .getItems
    .asScala.toSeq)
 }
}

private[spark] trait ExecutorPodsSnapshotsStore {

  def addSubscriber
      (processBatchIntervalMillis: Long)
      (onNewSnapshots: Seq[ExecutorPodsSnapshot] => Unit): Unit

  def stop(): Unit

  def notifySubscribers(): Unit

  def updatePod(updatedPod: Pod): Unit

  def replaceSnapshot(newSnapshot: Seq[Pod]): Unit
}

if (knownPodCount > targetNum) {
     val excess = knownPodCount - targetNum
     val knownPendingToDelete = currentPendingExecutors
       .filter(x => isExecutorIdleTimedOut(x.\_2, currentTime))
       .map { case (id, \_) => id }
       .take(excess - newlyCreatedExecutorsForRpId.size)
     val toDelete = newlyCreatedExecutorsForRpId
       .filter { case (\_, (\_, createTime)) =>
         currentTime - createTime > executorIdleTimeout
       }.keys.take(excess).toList ++ knownPendingToDelete

    if (toDelete.nonEmpty) {
      logInfo(s"Deleting ${toDelete.size} excess pod requests (${toDelete.mkString(",")}).")
      _deletedExecutorIds = _deletedExecutorIds ++ toDelete

      Utils.tryLogNonFatalError {

        kubernetesClient
          .pods()
          .withField("status.phase", "Pending")
          .withLabel(SPARK_APP_ID_LABEL, applicationId)
          .withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
          .withLabelIn(SPARK_EXECUTOR_ID_LABEL, toDelete.sorted.map(_.toString): _*)
          .delete()
        newlyCreatedExecutors --= toDelete
        knownPendingCount -= knownPendingToDelete.size
      }
    }
  }

  #创建ExecutorPod的逻辑

  for ( _ <- 0 until numExecutorsToAllocate) {
  val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
  val executorConf = KubernetesConf.createExecutorConf(
    conf,
    newExecutorId.toString,
    applicationId,
    driverPod,
    resourceProfileId)

  val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf, secMgr,
    kubernetesClient, rpIdToResourceProfile(resourceProfileId))
  val executorPod = resolvedExecutorSpec.pod
  val podWithAttachedContainer = new PodBuilder(executorPod.pod)
    .editOrNewSpec()
    .addToContainers(executorPod.container)
    .endSpec()
    .build()
  val createdExecutorPod = kubernetesClient.pods().create(podWithAttachedContainer)

schedulerBackend.doRemoveExecutor(execId.toString, exitReason)

     driverEndpoint.send(RemoveExecutor(executorId, reason))

kubernetesClient
.pods()
.withName(updatedPod.getMetadata.getName)
.delete()

#创建SchedulerBackend
createSchedulerBackend(){
        new KubernetesClusterSchedulerBackend(
      scheduler.asInstanceOf[TaskSchedulerImpl],
      sc,
      kubernetesClient,
      schedulerExecutorService,
      snapshotsStore,
      executorPodsAllocator,
      executorPodsLifecycleEventHandler,
      podsWatchEventSource,
      podsPollingEventSource)
}

  #初始化TaskScheduler，当taskScheduler运行起来后，这套消息系统也就自动运转起来
  override def initialize(scheduler: TaskScheduler, backend: SchedulerBackend): Unit = {
    scheduler.asInstanceOf[TaskSchedulerImpl].initialize(backend)
  }

override def start(): Unit = {
    super.start()
    val initExecs = Map(defaultProfile -> initialExecutors)
    podAllocator.setTotalExpectedExecutors(initExecs)
    lifecycleEventHandler.start(this)
    podAllocator.start(applicationId())
    watchEvents.start(applicationId())
    pollEvents.start(applicationId())
    setUpExecutorConfigMap()
  }