object SparkSubmit {
  def main(args: Array[String]): Unit = {
    val appArgs = new SparkSubmitArguments(args)
    if (appArgs.verbose) {
      // scalastyle:off println
      printStream.println(appArgs)
      // scalastyle:on println
    }
	// .action 默认为submit
    appArgs.action match {
      case SparkSubmitAction.SUBMIT => submit(appArgs)
      case SparkSubmitAction.KILL => kill(appArgs)
      case SparkSubmitAction.REQUEST_STATUS => requestStatus(appArgs)
    }
  }
  private def submit(args: SparkSubmitArguments): Unit = {
	// 准备提交job的环境，为runMain 准备参数
    val (childArgs, childClasspath, sysProps, childMainClass) = prepareSubmitEnvironment(args)
    def doRunMain(): Unit = {
	  ...
      runMain(childArgs, childClasspath, sysProps, childMainClass, args.verbose)
    }
	...
    doRunMain()
  }
}

object SparkSubmit {
  private def runMain(childArgs: Seq[String],childClasspath: Seq[String],sysProps: Map[String, String],childMainClass: String,verbose: Boolean): Unit = {
	val loader = 决定classloader
	Thread.currentThread.setContextClassLoader(loader)
	for (jar <- childClasspath) {
		addJarToClasspath(jar, loader)
	}
	for ((key, value) <- sysProps) {
		System.setProperty(key, value)
	}
	var mainClass: Class[_] = null
	mainClass = Utils.classForName(childMainClass)
	val mainMethod = mainClass.getMethod("main", new Array[String](0).getClass)
	mainMethod.invoke(null, childArgs.toArray)
  }
}

// spark/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
private object Client extends Logging {
  def main(argStrings: Array[String]) {
    // Set an env variable indicating we are running in YARN mode.
    // Note that any env variable with the SPARK_ prefix gets propagated to all (remote) processes
    System.setProperty("SPARK_YARN_MODE", "true")
    val sparkConf = new SparkConf
    // SparkSubmit would use yarn cache to distribute files & jars in yarn mode,so remove them from sparkConf here for yarn mode.
    sparkConf.remove("spark.jars")
    sparkConf.remove("spark.files")
    val args = new ClientArguments(argStrings)
    new Client(args, sparkConf).run()
  }
  // 向ResourceManager提交申请，获取appID
  def run(): Unit = {
    this.appId = submitApplication()
	// Submit an application to the ResourceManager.If set spark.yarn.submit.waitAppCompletion to true, it will stay alive reporting the application's status until the application has exited for any reason.Otherwise, the client process will exit after submission.If the application finishes with a failed, killed, or undefined status, throw an appropriate SparkException.
	...
  }
  // Submit an application running our ApplicationMaster to the ResourceManager.
  def submitApplication(): ApplicationId = {
    var appId: ApplicationId = null
    launcherBackend.connect()
    yarnClient.init(yarnConf)
    yarnClient.start()
    logInfo("Requesting a new application from cluster with %d NodeManagers".format(yarnClient.getYarnClusterMetrics.  getNumNodeManagers))
    // Get a new application from our RM
    val newApp = yarnClient.createApplication()
    val newAppResponse = newApp.getNewApplicationResponse()
    appId = newAppResponse.getApplicationId()
    reportLauncherState(SparkAppHandle.State.SUBMITTED)
    launcherBackend.setAppId(appId.toString)
    new CallerContext("CLIENT", Option(appId.toString)).setCurrentContext()
    // Verify whether the cluster has enough resources for our AM
    verifyClusterResources(newAppResponse)
    // Set up the appropriate contexts to launch our AM
    val containerContext = createContainerLaunchContext(newAppResponse)				// 封装命令
    val appContext = createApplicationSubmissionContext(newApp, containerContext)
    // Finally, submit and monitor the application
    logInfo(s"Submitting application $appId to ResourceManager")
    yarnClient.submitApplication(appContext)										// 提交命令
    appId
  }
}

// spark/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
object ApplicationMaster extends Logging {
  private var master: ApplicationMaster = _
  def main(args: Array[String]): Unit = {
    SignalUtils.registerLogger(log)
    val amArgs = new ApplicationMasterArguments(args)
    SparkHadoopUtil.get.runAsSparkUser { () =>
      master = new ApplicationMaster(amArgs, new YarnRMClient)
      System.exit(master.run())
    }
  }
}

// spark/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
private[spark] class ApplicationMaster(args: ApplicationMasterArguments,client: YarnRMClient)extends Logging {
  private val sparkConf = new SparkConf()
  private val yarnConf: YarnConfiguration = SparkHadoopUtil.get.newConfiguration(sparkConf).asInstanceOf[YarnConfiguration]
  private val isClusterMode = args.userClass != null

  // 仅保留 cluster mode 下的代码
  final def run(): Int = {
    val appAttemptId = client.getAttemptId()
    var attemptID: Option[String] = None

    System.setProperty("spark.ui.port", "0")
    System.setProperty("spark.master", "yarn")
    System.setProperty("spark.submit.deployMode", "cluster")
    System.setProperty("spark.yarn.app.id", appAttemptId.getApplicationId().toString())
    attemptID = Option(appAttemptId.getAttemptId.toString)
    
    new CallerContext("APPMASTER",Option(appAttemptId.getApplicationId.toString), attemptID).setCurrentContext()
    logInfo("ApplicationAttemptId: " + appAttemptId)
    val fs = FileSystem.get(yarnConf)	// 创建HDFS文件系统
    runDriver(securityMgr)				    // Driver的执行
    exitCode
  }
  private def runDriver(securityMgr: SecurityManager): Unit = {
    addAmIpFilter()
    userClassThread = startUserApplication()	// 启动了用户类线程

    logInfo("Waiting for spark context initialization...")
    val sc = ThreadUtils.awaitResult(sparkContextPromise.future,Duration(totalWaitTime, TimeUnit.MILLISECONDS))
    rpcEnv = sc.env.rpcEnv
    val driverRef = runAMEndpoint(sc.getConf.get("spark.driver.host"),sc.getConf.get("spark.driver.port"),isClusterMode = true)
    registerAM(sc.getConf, rpcEnv, driverRef, sc.ui.map(_.appUIAddress).getOrElse(""),securityMgr) // 向RM注册AM，向RM申请资源
    userClassThread.join()
  }
  private def startUserApplication(): Thread = {
    logInfo("Starting the user application in a separate Thread")

    val classpath = Client.getUserClasspath(sparkConf)
    val urls = classpath.map { entry =>
      new URL("file:" + new File(entry.getPath()).getAbsolutePath())
    }
    val userClassLoader = xx
    var userArgs = args.userArgs
    val mainMethod = userClassLoader.loadClass(args.userClass).getMethod("main", classOf[Array[String]])

    val userThread = new Thread {
      override def run() {
        mainMethod.invoke(null, userArgs.toArray) // 用类加载器的方式来加载用户类的main方法，并且,为这个线程设置名称为"Driver"
        finish(FinalApplicationStatus.SUCCEEDED, ApplicationMaster.EXIT_SUCCESS)
      }
    }
    userThread.setContextClassLoader(userClassLoader)
    userThread.setName("Driver")
    userThread.start()
    userThread
  }
  private def registerAM(_sparkConf: SparkConf,_rpcEnv: RpcEnv,driverRef: RpcEndpointRef,uiAddress: String,securityMgr: SecurityManager) = {
    val appId = client.getAttemptId().getApplicationId().toString()
    val attemptId = client.getAttemptId().getAttemptId().toString()
   
    val driverUrl = RpcEndpointAddress(_sparkConf.get("spark.driver.host"),_sparkConf.get("spark.driver.port").toInt,CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString

    allocator = client.register(driverUrl,driverRef,yarnConf,_sparkConf,uiAddress,historyAddress,securityMgr,localResources)
    allocator.allocateResources()   //让RM为AM分配资源
    reporterThread = launchReporterThread()
  }
}

// spark/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
private[yarn] class YarnAllocator(driverUrl: String,driverRef: RpcEndpointRef,conf: YarnConfiguration,sparkConf: SparkConf,amClient: AMRMClient[ContainerRequest],...)extends Logging {
  def allocateResources(): Unit = synchronized {
    val allocateResponse = amClient.allocate(progressIndicator)
    val allocatedContainers = allocateResponse.getAllocatedContainers() // 获取容器
    if (allocatedContainers.size > 0) { // 获取容器的大小
      handleAllocatedContainers(allocatedContainers.asScala)  // 处理获取到的容器
    }
  }
  def handleAllocatedContainers(allocatedContainers: Seq[Container]): Unit = {
    val containersToUse = new ArrayBuffer[Container](allocatedContainers.size)
	  // 计算获取到的资源中有没有一个 host、 机架/rack 的
    // Match incoming requests by host
    val remainingAfterHostMatches = new ArrayBuffer[Container]
    for (allocatedContainer <- allocatedContainers) {...}
    // Match remaining by rack
    val remainingAfterRackMatches = new ArrayBuffer[Container]
    for (allocatedContainer <- remainingAfterHostMatches) {...}
    // Assign remaining that are neither node-local nor rack-local
    val remainingAfterOffRackMatches = new ArrayBuffer[Container]
    for (allocatedContainer <- remainingAfterRackMatches) {...}
    runAllocatedContainers(containersToUse)	// 在这个Container里面运行ExecutorBackend
    logInfo("Received %d containers from YARN, launching executors on %d of them.".format(allocatedContainers.size, containersToUse.size))
  }
  private def runAllocatedContainers(containersToUse: ArrayBuffer[Container]): Unit = {
    for (container <- containersToUse) {
      val executorId = executorIdCounter.toString
      if (numExecutorsRunning < targetNumExecutors) {
        launcherPool.execute(new Runnable {
          override def run(): Unit = {
            new ExecutorRunnable(Some(container),conf,sparkConf,driverUrl,executorId,...).run()
            updateInternalState()
          }
        })
      } 
    }
  }
}

private[yarn] class ExecutorRunnable(container: Option[Container],conf: YarnConfiguration,sparkConf: SparkConf,masterAddress: String,executorId: String,...) extends Logging {
  def run(): Unit = {
    logDebug("Starting Executor Container")
    nmClient = NMClient.createNMClient()
    nmClient.init(conf)
    nmClient.start()
    startContainer()
  }
  def startContainer(): java.util.Map[String, ByteBuffer] = {
    val ctx = Records.newRecord(classOf[ContainerLaunchContext]).asInstanceOf[ContainerLaunchContext]
    val env = prepareEnvironment().asJava

    ctx.setLocalResources(localResources.asJava)
    ctx.setEnvironment(env)

    val credentials = UserGroupInformation.getCurrentUser().getCredentials()
    val dob = new DataOutputBuffer()
    credentials.writeTokenStorageToStream(dob)
    ctx.setTokens(ByteBuffer.wrap(dob.getData()))
    val commands = prepareCommand()
    ctx.setCommands(commands.asJava)
    ctx.setApplicationACLs(YarnSparkHadoopUtil.getApplicationAclsForYarn(securityMgr).asJava)
    // Send the start request to the ContainerManager
    nmClient.startContainer(container.get, ctx)
  }
}

// spark/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
private[spark] object CoarseGrainedExecutorBackend extends Logging {
  // 主要是对参数的赋值
  def main(args: Array[String]) {
    var driverUrl: String = null
    var executorId: String = null
    var hostname: String = null
    var cores: Int = 0
    var appId: String = null
    var workerUrl: Option[String] = None
    val userClassPath = new mutable.ListBuffer[URL]()
    var argv = args.toList
    // 用argv 给上述变量赋值
    run(driverUrl, executorId, hostname, cores, appId, workerUrl, userClassPath)
    System.exit(0)
  }
  private def run(driverUrl: String,executorId: String,hostname: String,cores: Int,appId: String,workerUrl: Option[String],userClassPath: Seq[URL]) {
    SparkHadoopUtil.get.runAsSparkUser { () =>
      // Bootstrap to fetch the driver's Spark properties.
      val executorConf = new SparkConf
      val port = executorConf.getInt("spark.executor.port", 0)
      val fetcher = RpcEnv.create("driverPropsFetcher",hostname,port,executorConf,new SecurityManager(executorConf),clientMode = true)
      val driver = fetcher.setupEndpointRefByURI(driverUrl)
      val cfg = driver.askWithRetry[SparkAppConfig](RetrieveSparkAppConfig)
      val props = cfg.sparkProperties ++ Seq[(String, String)](("spark.app.id", appId))
      fetcher.shutdown()

      // Create SparkEnv using properties we fetched from the driver.
      val driverConf = new SparkConf()
      for ((key, value) <- props) {
        // this is required for SSL in standalone mode
        if (SparkConf.isExecutorStartupConf(key)) {
          driverConf.setIfMissing(key, value)
        } else {
          driverConf.set(key, value)
        }
      }
      val env = SparkEnv.createExecutorEnv(driverConf, executorId, hostname, port, cores, cfg.ioEncryptionKey, isLocal = false)
      env.rpcEnv.setupEndpoint("Executor", new CoarseGrainedExecutorBackend(env.rpcEnv, driverUrl, executorId, hostname, cores, userClassPath, env))
      workerUrl.foreach { url =>
        env.rpcEnv.setupEndpoint("WorkerWatcher", new WorkerWatcher(env.rpcEnv, url))
      }
      env.rpcEnv.awaitTermination()
      SparkHadoopUtil.get.stopCredentialUpdater()
    }
  }
}

// spark/core/src/main/scala/org/apache/spark/rdd/RDD.scala
abstract class RDD[T: ClassTag]( @transient private var _sc: SparkContext, @transient private var deps: Seq[Dependency[_]]) extends Serializable with Logging {
  def collect(): Array[T] = withScope {
    val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
    Array.concat(results: _*)
  }
}
// spark/core/src/main/scala/org/apache/spark/SparkContext.scala
class SparkContext(config: SparkConf) extends Logging {
  def runJob[T, U: ClassTag](rdd: RDD[T],func: (TaskContext, Iterator[T]) => U,partitions: Seq[Int],resultHandler: (Int, U) => Unit): Unit = {
    val callSite = getCallSite
    val cleanedFunc = clean(func)
    logInfo("Starting job: " + callSite.shortForm)
    dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
    progressBar.foreach(_.finishAll())
    rdd.doCheckpoint()
  }
}
// spark/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
private[spark]
class DAGScheduler(private[scheduler] val sc: SparkContext,private[scheduler] val taskScheduler: TaskScheduler,env: SparkEnv,...)extends Logging {
  // Run an action job on the given RDD and pass all the results to the resultHandler function as they arrive.
  def runJob[T, U](rdd: RDD[T],func: (TaskContext, Iterator[T]) => U,partitions: Seq[Int],callSite: CallSite,resultHandler: (Int, U) => Unit,properties: Properties): Unit = {
    val start = System.nanoTime
    val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
    val awaitPermission = null.asInstanceOf[scala.concurrent.CanAwait]
    waiter.completionFuture.ready(Duration.Inf)(awaitPermission)
    waiter.completionFuture.value.get match {
      case scala.util.Success(_) =>
        logInfo("Job %d finished: %s, took %f s".format(waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
      case scala.util.Failure(exception) =>
        logInfo("Job %d failed: %s, took %f s".format(waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
        ...
    }
  }
  // Submit an action job to the scheduler.
  def submitJob[T, U](rdd: RDD[T],func: (TaskContext, Iterator[T]) => U,partitions: Seq[Int],callSite: CallSite,resultHandler: (Int, U) => Unit,properties: Properties): JobWaiter[U] = {
    val jobId = nextJobId.getAndIncrement()
    val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
    val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
    eventProcessLoop.post(JobSubmitted(jobId, rdd, func2, partitions.toArray, callSite, waiter,SerializationUtils.clone(properties))) // 发送一个提交任务的作业 ==> eventProcessLoop.onReceive ==> eventProcessLoop.doOnReceive ==> dagScheduler.handleJobSubmitted
    waiter
  }
  private[scheduler] def handleJobSubmitted(jobId: Int,finalRDD: RDD[_],func: (TaskContext, Iterator[_]) => _,partitions: Array[Int],callSite: CallSite,listener: JobListener,properties: Properties) {  
    var finalStage: ResultStage = null
    // New stage creation may throw an exception if, for example, jobs are run on a HadoopRDD whose underlying HDFS files have been deleted.
    finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite) // 先创建最终stage
    val job = new ActiveJob(jobId, finalStage, callSite, listener, properties)
    clearCacheLocs()
    logInfo("Got job %s (%s) with %d output partitions".format(job.jobId, callSite.shortForm, partitions.length))
    logInfo("Final stage: " + finalStage + " (" + finalStage.name + ")")
    logInfo("Parents of final stage: " + finalStage.parents)
    logInfo("Missing parents: " + getMissingParentStages(finalStage))

    val jobSubmissionTime = clock.getTimeMillis()
    jobIdToActiveJob(jobId) = job
    activeJobs += job
    finalStage.setActiveJob(job)
    val stageIds = jobIdToStageIds(jobId).toArray
    val stageInfos = stageIds.flatMap(id => stageIdToStage.get(id).map(_.latestInfo))
    listenerBus.post(SparkListenerJobStart(job.jobId, jobSubmissionTime, stageInfos, properties))
    submitStage(finalStage)   // 提交任务的最终stage
  }
  // Submits stage, but first recursively submits any missing parents.
  private def submitStage(stage: Stage) {
    val jobId = activeJobForStage(stage)
    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
      }
    }
  }
}