# Run on a Spark standalone cluster

./bin/spark-submit \

  --class org.apache.spark.examples.SparkPi \

  --master spark://207.184.161.138:7077 \

  --executor-memory 20G \

  --total-executor-cores  \

  /path/to/examples.jar \
  

[cp011@CDH- bin]$

more opt/cloudera/parcels/SPARK2-2.4..cloudera1-.cdh5.13.3.p0./lib/spark2/bin/spark-submit

#!/usr/bin/env bash

#

# Licensed to the Apache Software Foundation (ASF) under one or more

# contributor license agreements.  See the NOTICE file distributed with

# this work for additional information regarding copyright ownership.

# The ASF licenses this file to You under the Apache License, Version 2.0

# (the "License"); you may not use this file except in compliance with

# the License.  You may obtain a copy of the License at

#

#    http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

#

if [ -z "${SPARK_HOME}" ]; then

  source "$(dirname "$")"/find-spark-home

fi

# disable randomized hash for string in Python 3.3+

export PYTHONHASHSEED=

exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@"

object SparkSubmit extends CommandLineUtils with Logging {

  // Cluster managers

  private val YARN = 1

  private val STANDALONE = 2

  private val MESOS = 4

  private val LOCAL = 8

  private val KUBERNETES = 16

  private val ALL_CLUSTER_MGRS = YARN | STANDALONE | MESOS | LOCAL | KUBERNETES

  // Deploy modes

  private val CLIENT = 1

  private val CLUSTER = 2

  private val ALL_DEPLOY_MODES = CLIENT | CLUSTER

  // Special primary resource names that represent shells rather than application jars.

  private val SPARK_SHELL = "spark-shell"

  private val PYSPARK_SHELL = "pyspark-shell"

  private val SPARKR_SHELL = "sparkr-shell"

  private val SPARKR_PACKAGE_ARCHIVE = "sparkr.zip"

  private val R_PACKAGE_ARCHIVE = "rpkg.zip"

  private val CLASS_NOT_FOUND_EXIT_STATUS = 101

  // Following constants are visible for testing.

  private[deploy] val YARN_CLUSTER_SUBMIT_CLASS =

    "org.apache.spark.deploy.yarn.YarnClusterApplication"

  private[deploy] val REST_CLUSTER_SUBMIT_CLASS = classOf[RestSubmissionClientApp].getName()

  private[deploy] val STANDALONE_CLUSTER_SUBMIT_CLASS = classOf[ClientApp].getName()

  private[deploy] val KUBERNETES_CLUSTER_SUBMIT_CLASS =

    "org.apache.spark.deploy.k8s.submit.KubernetesClientApplication"

  override def main(args: Array[String]): Unit = {

    val submit = new SparkSubmit() {

      self =>

      override protected def parseArguments(args: Array[String]): SparkSubmitArguments = {

        new SparkSubmitArguments(args) {

          override protected def logInfo(msg: => String): Unit = self.logInfo(msg)

          override protected def logWarning(msg: => String): Unit = self.logWarning(msg)

        }

      }

      override protected def logInfo(msg: => String): Unit = printMessage(msg)

      override protected def logWarning(msg: => String): Unit = printMessage(s"Warning: $msg")

      override def doSubmit(args: Array[String]): Unit = {

        try {

          super.doSubmit(args)

        } catch {

          case e: SparkUserAppException =>

            exitFn(e.exitCode)

        }

      }

    }

    submit.doSubmit(args)

  }

  。。。

}

  def doSubmit(args: Array[String]): Unit = {

    // Initialize logging if it hasn't been done yet. Keep track of whether logging needs to

    // be reset before the application starts.

    val uninitLog = initializeLogIfNecessary(true, silent = true)

    val appArgs = parseArguments(args)

    if (appArgs.verbose) {

      logInfo(appArgs.toString)

    }

    appArgs.action match {

      case SparkSubmitAction.SUBMIT => submit(appArgs, uninitLog)

      case SparkSubmitAction.KILL => kill(appArgs)

      case SparkSubmitAction.REQUEST_STATUS => requestStatus(appArgs)

      case SparkSubmitAction.PRINT_VERSION => printVersion()

    }

  }

/**

   * 运行包含两步：

   * 第一步，我们通过设置适当的类路径，系统属性和应用程序参数来准备启动环境，以便基于集群管理和部署模式运行子主类。

   * 第二步，我们使用这个启动环境来调用子主类的主方法。

   * Submit the application using the provided parameters.

   * 使用提供的参数信息来提交application

   * This runs in two steps. First, we prepare the launch environment by setting up

   * the appropriate classpath, system properties, and application arguments for

   * running the child main class based on the cluster manager and the deploy mode.

   * Second, we use this launch environment to invoke the main method of the child

   * main class.

   */

  @tailrec

  private def submit(args: SparkSubmitArguments, uninitLog: Boolean): Unit = {

    // 通过设置适当的类路径，系统属性和应用程序参数来准备启动环境，以便基于集群管理和部署模式运行子主类。

    val (childArgs, childClasspath, sparkConf, childMainClass) = prepareSubmitEnvironment(args)

    def doRunMain(): Unit = {

      if (args.proxyUser != null) {

        val proxyUser = UserGroupInformation.createProxyUser(args.proxyUser,

          UserGroupInformation.getCurrentUser())

        try {

          proxyUser.doAs(new PrivilegedExceptionAction[Unit]() {

            override def run(): Unit = {

              runMain(childArgs, childClasspath, sparkConf, childMainClass, args.verbose)

            }

          })

        } catch {

          case e: Exception =>

            // Hadoop's AuthorizationException suppresses the exception's stack trace, which

            // makes the message printed to the output by the JVM not very helpful. Instead,

            // detect exceptions with empty stack traces here, and treat them differently.

            if (e.getStackTrace().length == 0) {

              error(s"ERROR: ${e.getClass().getName()}: ${e.getMessage()}")

            } else {

              throw e

            }

        }

      } else {

        runMain(childArgs, childClasspath, sparkConf, childMainClass, args.verbose)

      }

    }

    // Let the main class re-initialize the logging system once it starts.

    if (uninitLog) {

      Logging.uninitialize()

}

    //在独立集群模式下，有两个提交网关：

    //（1）使用o.a.s.deploy.Client作为包装器的传统RPC网关

    //（2）Spark 1.3中引入了新的基于REST的网关

    //后者是Spark 1.3的默认行为，但如果主端点不是REST服务器，则Spark Submit将故障转移到使用旧网关。

    // In standalone cluster mode, there are two submission gateways:

    //   (1) The traditional RPC gateway using o.a.s.deploy.Client as a wrapper

    //   (2) The new REST-based gateway introduced in Spark 1.3

    // The latter is the default behavior as of Spark 1.3, but Spark submit will fail over

    // to use the legacy gateway if the master endpoint turns out to be not a REST server.

    if (args.isStandaloneCluster && args.useRest) {

      try {

        logInfo("Running Spark using the REST application submission protocol.")

        doRunMain()

      } catch {

        // Fail over to use the legacy submission gateway

        case e: SubmitRestConnectionException =>

          logWarning(s"Master endpoint ${args.master} was not a REST server. " +

            "Falling back to legacy submission gateway instead.")

          args.useRest = false

          submit(args, false)

      }

    // 其他模式，只需直接运行主类

    // In all other modes, just run the main class as prepared

    } else {

      doRunMain()

    }

  }

val (childArgs, childClasspath, sparkConf, childMainClass) = prepareSubmitEnvironment(args)

/**

   * 未提交的应用程序准备环境

   * Prepare the environment for submitting an application.

   *

   * @param args the parsed SparkSubmitArguments used for environment preparation.

   * @param conf the Hadoop Configuration, this argument will only be set in unit test.

   * @return a 4-tuple:

   *        (1) the arguments for the child process,

   *        (2) a list of classpath entries for the child,

   *        (3) a map of system properties, and

   *        (4) the main class for the child

   *        返回一个4元组(childArgs, childClasspath, sparkConf, childMainClass)

   *        childArgs：子进程的参数

   *        childClasspath：子级的类路径条目列表

   *        sparkConf：系统参数map集合

   *        childMainClass：子级的主类

   *

   * Exposed for testing.

   */

  private[deploy] def prepareSubmitEnvironment(

      args: SparkSubmitArguments,

      conf: Option[HadoopConfiguration] = None)

      : (Seq[String], Seq[String], SparkConf, String) = {

    // Return values

    val childArgs = new ArrayBuffer[String]()

    val childClasspath = new ArrayBuffer[String]()

    val sparkConf = new SparkConf()

    var childMainClass = ""

    // 设置集群管理器，

    // 从这个列表中可以得到信息：spark目前支持的集群管理器包含：YARN,STANDLONE,MESOS,KUBERNETES,LOCAL，

    // 在spark-submit参数的--master中指定。

    // Set the cluster manager

    val clusterManager: Int = args.master match {

      case "yarn" => YARN

      case "yarn-client" | "yarn-cluster" =>

      // spark2.0之前可以使用yarn-cleint,yarn-cluster作为--master参数，从spark2.0起，不再支持，这里默认自动转化为yarn，并给出警告信息。

        logWarning(s"Master ${args.master} is deprecated since 2.0." +

          " Please use master \"yarn\" with specified deploy mode instead.")

        YARN

      case m if m.startsWith("spark") => STANDALONE

      case m if m.startsWith("mesos") => MESOS

      case m if m.startsWith("k8s") => KUBERNETES

      case m if m.startsWith("local") => LOCAL

      case _ =>

        error("Master must either be yarn or start with spark, mesos, k8s, or local")

        -1

    }

    // 设置部署模式--deploy-mode，默认为client模式。

    // Set the deploy mode; default is client mode

    var deployMode: Int = args.deployMode match {

      case "client" | null => CLIENT

      case "cluster" => CLUSTER

      case _ =>

        error("Deploy mode must be either client or cluster")

        -1

    }

    // 由于”yarn-cluster“和”yarn-client“方式已被弃用，因此封装了--master和--deploy-mode。

    // 如果只指定了一个--master和--deploy-mode，我们有一些逻辑来推断它们之间的关系；如果它们不一致，我们可以提前退出。

    // Because the deprecated way of specifying "yarn-cluster" and "yarn-client" encapsulate both

    // the master and deploy mode, we have some logic to infer the master and deploy mode

    // from each other if only one is specified, or exit early if they are at odds.

    if (clusterManager == YARN) {

      (args.master, args.deployMode) match {

        case ("yarn-cluster", null) =>

          deployMode = CLUSTER

          args.master = "yarn"

        case ("yarn-cluster", "client") =>

          error("Client deploy mode is not compatible with master \"yarn-cluster\"")

        case ("yarn-client", "cluster") =>

          error("Cluster deploy mode is not compatible with master \"yarn-client\"")

        case (_, mode) =>

          args.master = "yarn"

      }

      // 如果我们想去使用YARN的话，必须确保它包含在我们产品中。

      // Make sure YARN is included in our build if we're trying to use it

      if (!Utils.classIsLoadable(YARN_CLUSTER_SUBMIT_CLASS) && !Utils.isTesting) {

        error(

          "Could not load YARN classes. " +

          "This copy of Spark may not have been compiled with YARN support.")

      }

    }

    if (clusterManager == KUBERNETES) {

      args.master = Utils.checkAndGetK8sMasterUrl(args.master)

      // Make sure KUBERNETES is included in our build if we're trying to use it

      if (!Utils.classIsLoadable(KUBERNETES_CLUSTER_SUBMIT_CLASS) && !Utils.isTesting) {

        error(

          "Could not load KUBERNETES classes. " +

            "This copy of Spark may not have been compiled with KUBERNETES support.")

      }

    }

    // 下边的一些模式是不支持，尽早让它们失败。

    // Fail fast, the following modes are not supported or applicable

    (clusterManager, deployMode) match {

      case (STANDALONE, CLUSTER) if args.isPython =>

        error("Cluster deploy mode is currently not supported for python " +

          "applications on standalone clusters.")

      case (STANDALONE, CLUSTER) if args.isR =>

        error("Cluster deploy mode is currently not supported for R " +

          "applications on standalone clusters.")

      case (LOCAL, CLUSTER) =>

        error("Cluster deploy mode is not compatible with master \"local\"")

      case (_, CLUSTER) if isShell(args.primaryResource) =>

        error("Cluster deploy mode is not applicable to Spark shells.")

      case (_, CLUSTER) if isSqlShell(args.mainClass) =>

        error("Cluster deploy mode is not applicable to Spark SQL shell.")

      case (_, CLUSTER) if isThriftServer(args.mainClass) =>

        error("Cluster deploy mode is not applicable to Spark Thrift server.")

      case _ =>

    }

    // 如果args.deployMode为null的话，给它赋值更新。稍后它将作为Spark的属性向下传递

    // Update args.deployMode if it is null. It will be passed down as a Spark property later.

    (args.deployMode, deployMode) match {

      case (null, CLIENT) => args.deployMode = "client"

      case (null, CLUSTER) => args.deployMode = "cluster"

      case _ =>

    }

    // 根据资源管理器和部署模式，进行逻辑判断出几种特殊运行方式。

    val isYarnCluster = clusterManager == YARN && deployMode == CLUSTER

    val isMesosCluster = clusterManager == MESOS && deployMode == CLUSTER

    val isStandAloneCluster = clusterManager == STANDALONE && deployMode == CLUSTER

    val isKubernetesCluster = clusterManager == KUBERNETES && deployMode == CLUSTER

    val isMesosClient = clusterManager == MESOS && deployMode == CLIENT

    if (!isMesosCluster && !isStandAloneCluster) {

      // Resolve maven dependencies if there are any and add classpath to jars. Add them to py-files

      // too for packages that include Python code

      val resolvedMavenCoordinates = DependencyUtils.resolveMavenDependencies(

        args.packagesExclusions, args.packages, args.repositories, args.ivyRepoPath,

        args.ivySettingsPath)

      if (!StringUtils.isBlank(resolvedMavenCoordinates)) {

        args.jars = mergeFileLists(args.jars, resolvedMavenCoordinates)

        if (args.isPython || isInternal(args.primaryResource)) {

          args.pyFiles = mergeFileLists(args.pyFiles, resolvedMavenCoordinates)

        }

      }

      // install any R packages that may have been passed through --jars or --packages.

      // Spark Packages may contain R source code inside the jar.

      if (args.isR && !StringUtils.isBlank(args.jars)) {

        RPackageUtils.checkAndBuildRPackage(args.jars, printStream, args.verbose)

      }

    }

    args.sparkProperties.foreach { case (k, v) => sparkConf.set(k, v) }

    val hadoopConf = conf.getOrElse(SparkHadoopUtil.newConfiguration(sparkConf))

    val targetDir = Utils.createTempDir()

    // assure a keytab is available from any place in a JVM

    if (clusterManager == YARN || clusterManager == LOCAL || isMesosClient) {

      if (args.principal != null) {

        if (args.keytab != null) {

          require(new File(args.keytab).exists(), s"Keytab file: ${args.keytab} does not exist")

          // Add keytab and principal configurations in sysProps to make them available

          // for later use; e.g. in spark sql, the isolated class loader used to talk

          // to HiveMetastore will use these settings. They will be set as Java system

          // properties and then loaded by SparkConf

          sparkConf.set(KEYTAB, args.keytab)

          sparkConf.set(PRINCIPAL, args.principal)

          UserGroupInformation.loginUserFromKeytab(args.principal, args.keytab)

        }

      }

    }

    // Resolve glob path for different resources.

    args.jars = Option(args.jars).map(resolveGlobPaths(_, hadoopConf)).orNull

    args.files = Option(args.files).map(resolveGlobPaths(_, hadoopConf)).orNull

    args.pyFiles = Option(args.pyFiles).map(resolveGlobPaths(_, hadoopConf)).orNull

    args.archives = Option(args.archives).map(resolveGlobPaths(_, hadoopConf)).orNull

    lazy val secMgr = new SecurityManager(sparkConf)

    // In client mode, download remote files.

    var localPrimaryResource: String = null

    var localJars: String = null

    var localPyFiles: String = null

    if (deployMode == CLIENT) {

      localPrimaryResource = Option(args.primaryResource).map {

        downloadFile(_, targetDir, sparkConf, hadoopConf, secMgr)

      }.orNull

      localJars = Option(args.jars).map {

        downloadFileList(_, targetDir, sparkConf, hadoopConf, secMgr)

      }.orNull

      localPyFiles = Option(args.pyFiles).map {

        downloadFileList(_, targetDir, sparkConf, hadoopConf, secMgr)

      }.orNull

    }

    // When running in YARN, for some remote resources with scheme:

    //   1. Hadoop FileSystem doesn't support them.

    //   2. We explicitly bypass Hadoop FileSystem with "spark.yarn.dist.forceDownloadSchemes".

    // We will download them to local disk prior to add to YARN's distributed cache.

    // For yarn client mode, since we already download them with above code, so we only need to

    // figure out the local path and replace the remote one.

    if (clusterManager == YARN) {

      val forceDownloadSchemes = sparkConf.get(FORCE_DOWNLOAD_SCHEMES)

      def shouldDownload(scheme: String): Boolean = {

        forceDownloadSchemes.contains("*") || forceDownloadSchemes.contains(scheme) ||

          Try { FileSystem.getFileSystemClass(scheme, hadoopConf) }.isFailure

      }

      def downloadResource(resource: String): String = {

        val uri = Utils.resolveURI(resource)

        uri.getScheme match {

          case "local" | "file" => resource

          case e if shouldDownload(e) =>

            val file = new File(targetDir, new Path(uri).getName)

            if (file.exists()) {

              file.toURI.toString

            } else {

              downloadFile(resource, targetDir, sparkConf, hadoopConf, secMgr)

            }

          case _ => uri.toString

        }

      }

      args.primaryResource = Option(args.primaryResource).map { downloadResource }.orNull

      args.files = Option(args.files).map { files =>

        Utils.stringToSeq(files).map(downloadResource).mkString(",")

      }.orNull

      args.pyFiles = Option(args.pyFiles).map { pyFiles =>

        Utils.stringToSeq(pyFiles).map(downloadResource).mkString(",")

      }.orNull

      args.jars = Option(args.jars).map { jars =>

        Utils.stringToSeq(jars).map(downloadResource).mkString(",")

      }.orNull

      args.archives = Option(args.archives).map { archives =>

        Utils.stringToSeq(archives).map(downloadResource).mkString(",")

      }.orNull

    }

    // If we're running a python app, set the main class to our specific python runner

    。。。。

    // In YARN mode for an R app, add the SparkR package archive and the R package

    // archive containing all of the built R libraries to archives so that they can

    // be distributed with the job

    。。。。

    // TODO: Support distributing R packages with standalone cluster

    。。。。

    // TODO: Support distributing R packages with mesos cluster

    。。。。

    // If we're running an R app, set the main class to our specific R runner

    。。。。   

    // Special flag to avoid deprecation warnings at the client

    sys.props("SPARK_SUBMIT") = "true"

    // In client mode, launch the application main class directly

    // In addition, add the main application jar and any added jars (if any) to the classpath

    if (deployMode == CLIENT) {

      childMainClass = args.mainClass

      if (localPrimaryResource != null && isUserJar(localPrimaryResource)) {

        childClasspath += localPrimaryResource

      }

      if (localJars != null) { childClasspath ++= localJars.split(",") }

    }

    // Add the main application jar and any added jars to classpath in case YARN client

    // requires these jars.

    // This assumes both primaryResource and user jars are local jars, or already downloaded

    // to local by configuring "spark.yarn.dist.forceDownloadSchemes", otherwise it will not be

    // added to the classpath of YARN client.

    if (isYarnCluster) {

      if (isUserJar(args.primaryResource)) {

        childClasspath += args.primaryResource

      }

      if (args.jars != null) { childClasspath ++= args.jars.split(",") }

    }

    if (deployMode == CLIENT) {

      if (args.childArgs != null) { childArgs ++= args.childArgs }

    }

    // Map all arguments to command-line options or system properties for our chosen mode

    for (opt <- options) {

      if (opt.value != null &&

          (deployMode & opt.deployMode) != 0 &&

          (clusterManager & opt.clusterManager) != 0) {

        if (opt.clOption != null) { childArgs += (opt.clOption, opt.value) }

        if (opt.confKey != null) { sparkConf.set(opt.confKey, opt.value) }

      }

    }

    // In case of shells, spark.ui.showConsoleProgress can be true by default or by user.

    if (isShell(args.primaryResource) && !sparkConf.contains(UI_SHOW_CONSOLE_PROGRESS)) {

      sparkConf.set(UI_SHOW_CONSOLE_PROGRESS, true)

    }

    // Let YARN know it's a pyspark app, so it distributes needed libraries.

    if (clusterManager == YARN) {

      if (args.isPython) {

        sparkConf.set("spark.yarn.isPython", "true")

      }

    }

    // In yarn-cluster mode, use yarn.Client as a wrapper around the user class

    if (isYarnCluster) {

      childMainClass = YARN_CLUSTER_SUBMIT_CLASS

      if (args.isPython) {

        childArgs += ("--primary-py-file", args.primaryResource)

        childArgs += ("--class", "org.apache.spark.deploy.PythonRunner")

      } else if (args.isR) {

        val mainFile = new Path(args.primaryResource).getName

        childArgs += ("--primary-r-file", mainFile)

        childArgs += ("--class", "org.apache.spark.deploy.RRunner")

      } else {

        if (args.primaryResource != SparkLauncher.NO_RESOURCE) {

          childArgs += ("--jar", args.primaryResource)

        }

        childArgs += ("--class", args.mainClass)

      }

      if (args.childArgs != null) {

        args.childArgs.foreach { arg => childArgs += ("--arg", arg) }

      }

    }

    // Load any properties specified through --conf and the default properties file

    for ((k, v) <- args.sparkProperties) {

      sparkConf.setIfMissing(k, v)

    }

    // Ignore invalid spark.driver.host in cluster modes.

    if (deployMode == CLUSTER) {

      sparkConf.remove("spark.driver.host")

    }

    // Resolve paths in certain spark properties

    val pathConfigs = Seq(

      "spark.jars",

      "spark.files",

      "spark.yarn.dist.files",

      "spark.yarn.dist.archives",

      "spark.yarn.dist.jars")

    pathConfigs.foreach { config =>

      // Replace old URIs with resolved URIs, if they exist

      sparkConf.getOption(config).foreach { oldValue =>

        sparkConf.set(config, Utils.resolveURIs(oldValue))

      }

    }

    // Resolve and format python file paths properly before adding them to the PYTHONPATH.

    // The resolving part is redundant in the case of --py-files, but necessary if the user

    // explicitly sets `spark.submit.pyFiles` in his/her default properties file.

    sparkConf.getOption("spark.submit.pyFiles").foreach { pyFiles =>

      val resolvedPyFiles = Utils.resolveURIs(pyFiles)

      val formattedPyFiles = if (!isYarnCluster && !isMesosCluster) {

        PythonRunner.formatPaths(resolvedPyFiles).mkString(",")

      } else {

        // Ignoring formatting python path in yarn and mesos cluster mode, these two modes

        // support dealing with remote python files, they could distribute and add python files

        // locally.

        resolvedPyFiles

      }

      sparkConf.set("spark.submit.pyFiles", formattedPyFiles)

    }

    (childArgs, childClasspath, sparkConf, childMainClass)

  }

private[deploy] def prepareSubmitEnvironment(args: SparkSubmitArguments,conf: Option[HadoopConfiguration] = None): (Seq[String], Seq[String], SparkConf, String)

private[spark] class YarnClusterApplication extends SparkApplication {

  override def start(args: Array[String], conf: SparkConf): Unit = {

    // SparkSubmit would use yarn cache to distribute files & jars in yarn mode,

    // so remove them from sparkConf here for yarn mode.

    conf.remove("spark.jars")

    conf.remove("spark.files")

    new Client(new ClientArguments(args), conf).run()

  }

}

/**

 * Entry point for a Spark application. Implementations must provide a no-argument constructor.

 */

private[spark] trait SparkApplication {

  def start(args: Array[String], conf: SparkConf): Unit

}

/**

 * Implementation of SparkApplication that wraps a standard Java class with a "main" method.

 *

 * Configuration is propagated to the application via system properties, so running multiple

 * of these in the same JVM may lead to undefined behavior due to configuration leaks.

 */

private[deploy] class JavaMainApplication(klass: Class[_]) extends SparkApplication {

  override def start(args: Array[String], conf: SparkConf): Unit = {

    val mainMethod = klass.getMethod("main", new Array[String](0).getClass)

    if (!Modifier.isStatic(mainMethod.getModifiers)) {

      throw new IllegalStateException("The main method in the given main class must be static")

    }

    val sysProps = conf.getAll.toMap

    sysProps.foreach { case (k, v) =>

      sys.props(k) = v

    }

    mainMethod.invoke(null, args)

  }

}

/**

   * 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.

   */

  def run(): Unit = {

    this.appId = submitApplication()

    if (!launcherBackend.isConnected() && fireAndForget) {

      val report = getApplicationReport(appId)

      val state = report.getYarnApplicationState

      logInfo(s"Application report for $appId (state: $state)")

      logInfo(formatReportDetails(report))

      if (state == YarnApplicationState.FAILED || state == YarnApplicationState.KILLED) {

        throw new SparkException(s"Application $appId finished with status: $state")

      }

    } else {

      val YarnAppReport(appState, finalState, diags) = monitorApplication(appId)

      if (appState == YarnApplicationState.FAILED || finalState == FinalApplicationStatus.FAILED) {

        diags.foreach { err =>

          logError(s"Application diagnostics message: $err")

        }

        throw new SparkException(s"Application $appId finished with failed status")

      }

      if (appState == YarnApplicationState.KILLED || finalState == FinalApplicationStatus.KILLED) {

        throw new SparkException(s"Application $appId is killed")

      }

      if (finalState == FinalApplicationStatus.UNDEFINED) {

        throw new SparkException(s"The final status of application $appId is undefined")

      }

    }

  }

  /**

   * Submit an application running our ApplicationMaster to the ResourceManager.

   *

   * The stable Yarn API provides a convenience method (YarnClient#createApplication) for

   * creating applications and setting up the application submission context. This was not

   * available in the alpha API.

   */

  def submitApplication(): ApplicationId = {

    var appId: ApplicationId = null

    try {

      launcherBackend.connect()

      yarnClient.init(hadoopConf)

      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()

      new CallerContext("CLIENT", sparkConf.get(APP_CALLER_CONTEXT),

        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)

      launcherBackend.setAppId(appId.toString)

      reportLauncherState(SparkAppHandle.State.SUBMITTED)

      appId

    } catch {

      case e: Throwable =>

        if (appId != null) {

          cleanupStagingDir(appId)

        }

        throw e

    }

  }

  private var _schedulerBackend: SchedulerBackend = _

  private var _taskScheduler: TaskScheduler = _

  private[spark] def schedulerBackend: SchedulerBackend = _schedulerBackend

  private[spark] def taskScheduler: TaskScheduler = _taskScheduler

  private[spark] def taskScheduler_=(ts: TaskScheduler): Unit = {

    _taskScheduler = ts

  }

  // 构造函数中初始化赋值

    // Create and start the scheduler

    val (sched, ts) = SparkContext.createTaskScheduler(this, master, deployMode)

    _schedulerBackend = sched

_taskScheduler = ts

/**

   * Create a task scheduler based on a given master URL.

   * Return a 2-tuple of the scheduler backend and the task scheduler.

   */

  private def createTaskScheduler(。。。): (SchedulerBackend, TaskScheduler) = {

    。。。

    master match {

      case "local" =>

        。。。

      case LOCAL_N_REGEX(threads) =>

        。。。

      case LOCAL_N_FAILURES_REGEX(threads, maxFailures) =>

        。。。。

      case SPARK_REGEX(sparkUrl) =>

        。。。。

      case LOCAL_CLUSTER_REGEX(numSlaves, coresPerSlave, memoryPerSlave) =>

        。。。。

      case masterUrl =>

        val cm = getClusterManager(masterUrl) match {

          case Some(clusterMgr) => clusterMgr

          case None => throw new SparkException("Could not parse Master URL: '" + master + "'")

        }

        try {

          val scheduler = cm.createTaskScheduler(sc, masterUrl)

          val backend = cm.createSchedulerBackend(sc, masterUrl, scheduler)

          cm.initialize(scheduler, backend)

          (backend, scheduler)

        } catch {

          case se: SparkException => throw se

          case NonFatal(e) =>

            throw new SparkException("External scheduler cannot be instantiated", e)

        }

    }

  }

  private def getClusterManager(url: String): Option[ExternalClusterManager] = {

    val loader = Utils.getContextOrSparkClassLoader

    val serviceLoaders =

      ServiceLoader.load(classOf[ExternalClusterManager], loader).asScala.filter(_.canCreate(url))

    if (serviceLoaders.size > 1) {

      throw new SparkException(

        s"Multiple external cluster managers registered for the url $url: $serviceLoaders")

    }

    serviceLoaders.headOption

  }  

private[spark] trait ExternalClusterManager {

  def canCreate(masterURL: String): Boolean

  def createTaskScheduler(sc: SparkContext, masterURL: String): TaskScheduler

  def createSchedulerBackend(sc: SparkContext,

      masterURL: String,

      scheduler: TaskScheduler): SchedulerBackend

  def initialize(scheduler: TaskScheduler, backend: SchedulerBackend): Unit

}

private[spark] class YarnClusterManager extends ExternalClusterManager {

  override def canCreate(masterURL: String): Boolean = {

    masterURL == "yarn"

  }

  override def createTaskScheduler(sc: SparkContext, masterURL: String): TaskScheduler = {

    sc.deployMode match {

      case "cluster" => new YarnClusterScheduler(sc)

      case "client" => new YarnScheduler(sc)

      case _ => throw new SparkException(s"Unknown deploy mode '${sc.deployMode}' for Yarn")

    }

  }

  override def createSchedulerBackend(sc: SparkContext,

      masterURL: String,

      scheduler: TaskScheduler): SchedulerBackend = {

    sc.deployMode match {

      case "cluster" =>

        new YarnClusterSchedulerBackend(scheduler.asInstanceOf[TaskSchedulerImpl], sc)

      case "client" =>

        new YarnClientSchedulerBackend(scheduler.asInstanceOf[TaskSchedulerImpl], sc)

      case  _ =>

        throw new SparkException(s"Unknown deploy mode '${sc.deployMode}' for Yarn")

    }

  }

  override def initialize(scheduler: TaskScheduler, backend: SchedulerBackend): Unit = {

    scheduler.asInstanceOf[TaskSchedulerImpl].initialize(backend)

  }

}

// Default to numExecutors * 2, with minimum of 3

private val maxNumExecutorFailures =sparkConf.getInt("spark.yarn.max.executor.failures",

sparkConf.getInt("spark.yarn.max.worker.failures", math.max(args.numExecutors *,)))