Spark源码分析 之 Driver和Excutor是怎么跑起来的?(2.2.0版本)

// 创建并启动调度器
val (sched, ts) = SparkContext.createTaskScheduler(this, master, deployMode)
_schedulerBackend = sched
_taskScheduler = ts
_dagScheduler = new DAGScheduler(this)
...
// 启动调度器
_taskScheduler.start()

master match {
      case "local" =>
        ...

      case LOCAL_N_REGEX(threads) =>
        ...

      case LOCAL_N_FAILURES_REGEX(threads, maxFailures) =>
        ...

      case SPARK_REGEX(sparkUrl) =>
        // 创建调度器
        val scheduler = new TaskSchedulerImpl(sc)
        val masterUrls = sparkUrl.split(",").map("spark://" + _)
        // 创建backend
        val backend = new StandaloneSchedulerBackend(scheduler, sc, masterUrls)
        // 把backend注入到schduler中
        scheduler.initialize(backend)
        (backend, scheduler)

      case LOCAL_CLUSTER_REGEX(numSlaves, coresPerSlave, memoryPerSlave) =>
        ...

      case masterUrl =>
        ...
    }

def initialize(backend: SchedulerBackend) {
    this.backend = backend
    schedulableBuilder = {
      schedulingMode match {
        case SchedulingMode.FIFO =>
          new FIFOSchedulableBuilder(rootPool)
        case SchedulingMode.FAIR =>
          new FairSchedulableBuilder(rootPool, conf)
        case _ =>
          throw new IllegalArgumentException(s"Unsupported $SCHEDULER_MODE_PROPERTY: " +
          s"$schedulingMode")
      }
    }
    schedulableBuilder.buildPools()
  }

override def start() {
    backend.start()
    ...
  }

override def start() {
    //初始化参数
    ...
    
    val command = Command("org.apache.spark.executor.CoarseGrainedExecutorBackend",
      args, sc.executorEnvs, classPathEntries ++ testingClassPath, libraryPathEntries, javaOpts)
   ...
    val appDesc = ApplicationDescription(sc.appName, maxCores, sc.executorMemory, command,
      webUrl, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor, initialExecutorLimit)
    // 注意前面创建了一大堆的配置对象，主要就是那个class等信息
    client = new StandaloneAppClient(sc.env.rpcEnv, masters, appDesc, this, conf)
    client.start()
    ...
  }

override def onStart(): Unit = {
  try {
    registerWithMaster(1)//发起注册
  } catch {
    ...
  }
}

private def registerWithMaster(nthRetry: Int) {
  registerMasterFutures.set(tryRegisterAllMasters())
  registrationRetryTimer.set(registrationRetryThread.schedule(new Runnable {
    override def run(): Unit = {
      if (registered.get) {
        registerMasterFutures.get.foreach(_.cancel(true))
        registerMasterThreadPool.shutdownNow()
      } else if (nthRetry >= REGISTRATION_RETRIES) {
        markDead("All masters are unresponsive! Giving up.")
      } else {
        registerMasterFutures.get.foreach(_.cancel(true))
        registerWithMaster(nthRetry + 1)
      }
    }
  }, REGISTRATION_TIMEOUT_SECONDS, TimeUnit.SECONDS))
}
//发起注册
private def tryRegisterAllMasters(): Array[JFuture[_]] = {
  ...
  masterRef.send(RegisterApplication(appDescription, self))
  ...
}

case RegisterApplication(description, driver) =>
   // TODO Prevent repeated registrations from some driver
   if (state == RecoveryState.STANDBY) {
     // ignore, don't send response
   } else {
     ...
     schedule()
   }

private def schedule(): Unit = {
  ...
  for (driver <- waitingDrivers.toList) { // iterate over a copy of waitingDrivers
    ...
    while (numWorkersVisited < numWorkersAlive && !launched) {
      ...
      if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
        launchDriver(worker, driver)
        ...
      }
      ...
    }
  }
  startExecutorsOnWorkers()
}
//向worker发送launchDriver请求
private def launchDriver(worker: WorkerInfo, driver: DriverInfo) {
  ...
  worker.endpoint.send(LaunchDriver(driver.id, driver.desc))
  ...
}

case LaunchDriver(driverId, driverDesc) =>
      logInfo(s"Asked to launch driver $driverId")
      val driver = new DriverRunner(
        conf,
        driverId,
        workDir,
        sparkHome,
        driverDesc.copy(command = Worker.maybeUpdateSSLSettings(driverDesc.command, conf)),
        self,
        workerUri,
        securityMgr)
      drivers(driverId) = driver
      driver.start()

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

private def startExecutorsOnWorkers(): Unit = {
...
  for (app <- waitingApps if app.coresLeft > 0) {
    ...
    for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) {
      allocateWorkerResourceToExecutors(
        app, assignedCores(pos), coresPerExecutor, usableWorkers(pos))
    }
  }
}

private def allocateWorkerResourceToExecutors(
   app: ApplicationInfo,
   assignedCores: Int,
   coresPerExecutor: Option[Int],
   worker: WorkerInfo): Unit = {
 ...
 for (i <- 1 to numExecutors) {
   ...
   launchExecutor(worker, exec)
   ...
 }
}

private def launchExecutor(worker: WorkerInfo, exec: ExecutorDesc): Unit = {
  ...
  worker.endpoint.send(LaunchExecutor(masterUrl,
    exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory))
  ...
}

case LaunchExecutor(masterUrl, appId, execId, appDesc, cores_, memory_) =>
if (masterUrl != activeMasterUrl) {
  ...
} else {
  try {
    ...
    val manager = new ExecutorRunner(
      appId,
      execId,
      appDesc.copy(command = Worker.maybeUpdateSSLSettings(appDesc.command, conf)),
      cores_,
      memory_,
      self,
      workerId,
      host,
      webUi.boundPort,
      publicAddress,
      sparkHome,
      executorDir,
      workerUri,
      conf,
      appLocalDirs, ExecutorState.RUNNING)
    ...
  } catch {
...
  }
}