java - spark如何向集群发送作业？

Question

我正在使用 YARN 并试图了解 Spark 如何使用 YARN 将作业发送到集群。因此，我深入研究了资源，发现当我们提交作业(例如 foreach)时，以下方法正在 SparkContext::runJob 中执行:

dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)

然后 JobWaiter 被创建在

DAGScheduler::submitJob 

并且正在发布 JobSubmitted 事件。该事件正在处理中

DAGSchedulerEventProcessLoop::handleJobSubmitted 

将另一个事件 (SparkListenerJobStart) 发布到监听器总线。然后调用

DAGScheduler::submitStage

所以在方法中似乎应该有将阶段提交到集群的逻辑。但我唯一看到的是:

  private def submitStage(stage: Stage) {
    val jobId = activeJobForStage(stage)
    if (jobId.isDefined) {
      logDebug("submitStage(" + stage + ")")
      if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
        val missing = getMissingParentStages(stage).sortBy(_.id)
        logDebug("missing: " + missing)
        if (missing.isEmpty) {
          logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
          submitMissingTasks(stage, jobId.get)
        } else {
          for (parent <- missing) {
            submitStage(parent)
          }
          waitingStages += stage
        }
      }
    } else {
      abortStage(stage, "No active job for stage " + stage.id, None)
    }
  }

实际上，我希望那里有一些类似 NIO 的代码。

问题:如果我们使用YARN，驱动程序如何与集群通信？这段代码在哪里？有人可以帮助我吗？

Answer 1

如您所知，Spark 可以在多个集群管理器上运行。 Spark 通过使用名为 SchedulerBackend 的抽象来实现这一点。

对于 YARN，有两种实现方式:

1. YarnClientSchedulerBackend(用于客户端部署模式)
2. YarnSchedulerBackend(用于集群部署模式)

这是源代码:https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/SchedulerBackend.scala