Spark源码 —— 从 SparkSubmit 到 Driver启动

  def main(args: Array[String]): Unit = {
    val appArgs = new SparkSubmitArguments(args)
    if (appArgs.verbose) {
      // scalastyle:off println
      printStream.println(appArgs)
      // scalastyle:on println
    }
    appArgs.action match {
      case SparkSubmitAction.SUBMIT => submit(appArgs)
      case SparkSubmitAction.KILL => kill(appArgs)
      case SparkSubmitAction.REQUEST_STATUS => requestStatus(appArgs)
    }
  }

···
private[deploy] def prepareSubmitEnvironment(args: 
SparkSubmitArguments)    : (Seq[String], Seq[String], Map[String, String], String) = {  // Return values  
val childArgs = new ArrayBuffer[String]() 
val childClasspath = new ArrayBuffer[String]()  
val sysProps = new HashMap[String, String]()  
var childMainClass = ""
 ...
 返回值
  (childArgs, childClasspath, sysProps, childMainClass)

 private def schedule(): Unit = {
    if (state != RecoveryState.ALIVE) { return }
    // 将worker打乱，主要就是为了负载均衡
    val shuffledWorkers = Random.shuffle(workers)
    //筛选存活的 worker
    for (worker <- shuffledWorkers if worker.state == WorkerState.ALIVE) {
      // 遍历等待启动的Driver
      for (driver <- waitingDrivers) {
        //如果该worker内存和core都满足要求
        if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
         // 启动Driver
          launchDriver(worker, driver)
          waitingDrivers -= driver
        }
      }
    }
    //实际上还会去启动Executor，
    //但是我们目前不关注这里，略过
    startExecutorsOnWorkers()
  }

  private def launchDriver(worker: WorkerInfo, driver: DriverInfo) {
   //将driver的信息记录下来
    worker.addDriver(driver)
    // driver现在知道他该在哪个worker启动了
    driver.worker = Some(worker)
    // 向worker节点发送 LaunchDriver
    worker.endpoint.send(LaunchDriver(driver.id, driver.desc))
    // 标记driver已经运行，实际是Driver可能还没启动呢！！！
    driver.state = DriverState.RUNNING
  }

//节选代码，不要介意是 case 开头
 case LaunchDriver(driverId, driverDesc) => {
      logInfo(s"Asked to launch driver $driverId")
      //new一个 DriverRunner 
      val driver = new DriverRunner(
        conf,
        driverId,
        workDir,
        sparkHome,
        driverDesc.copy(command = Worker.maybeUpdateSSLSettings(driverDesc.command, conf)),
        self,
        workerUri,
        securityMgr)
        
      //DriverRunner start
      drivers(driverId) = driver
      driver.start()
      //记录下消耗的资源
      coresUsed += driverDesc.cores
      memoryUsed += driverDesc.mem
    }

new Thread("DriverRunner for " + driverId) {
    override def run(){
        ...
       //创建driver工作目录
       val driverDir = createWorkingDirectory()
       //下载一些jars
       val localJarFilename = downloadUserJar(driverDir)
        ...
        // 通过系统的指令创建 jvm进程，至此正式启动
        launchDriver(builder, driverDir, driverDesc.supervise)
    }
}.start()

// Delegate to supplied main class
val clazz = Utils.classForName(mainClass)
val mainMethod = clazz.getMethod("main", classOf[Array[String]])
mainMethod.invoke(null, extraArgs.toArray[String])