/**   * @return a tuple containing   *           (1) the arguments for the child process,   *           (2) a list of classpath entries for the child,   *           (3) a list of system properties and env vars, and   *           (4) the main class for the child   *///createLaunchEnv的方法返回值//1.子进程的参数,字符串数组，ArrayBuffer[String]//2.子进程JVM的classpath路径列表,字符串数组，ArrayBuffer[String]//3.子进程的系统变量和环境变量 ，HashMap类型//4.子进程JVM的main classprivate[spark] def createLaunchEnv(args: SparkSubmitArguments)      : (ArrayBuffer[String], ArrayBuffer[String], Map[String, String], String) = {    // Values to return    val childArgs = new ArrayBuffer[String]()    val childClasspath = new ArrayBuffer[String]()    val sysProps = new HashMap[String, String]()    var childMainClass = ""    // Set the cluster manager    //集群管理器，这里指定了四种：YARN，STANDALONE，MESON和LOCAL    //需要注意的是，为什么LOCAL也是一种集群管理器，它的集群含义是什么？    //根据args.master参数值决定clusterManager，注意，区分大小写    //这里只检查master是否以yarn, spark, mesos或者local开头，实际中，以yarn开头的master值可能是yarn-client，yarn-cluster，yarn-standalone，所以代码后面对master做了更进一步的检查    val clusterManager: Int = args.master match {      case m if m.startsWith("yarn") => YARN      case m if m.startsWith("spark") => STANDALONE      case m if m.startsWith("mesos") => MESOS      case m if m.startsWith("local") => LOCAL      //如果master不以这四个开头，提示出错信息是***Master***必须以yarn, spark, mesos, or local开头      case _ => printErrorAndExit("Master must start with yarn, spark, mesos, or local"); -1    }    // Set the deploy mode; default is client mode    //设置部署模式，有两种模式，client和cluster模式    //如果没有设置deployMode(取值null)，则默认认为是client模式    var deployMode: Int = args.deployMode match {      case "client" | null => CLIENT      case "cluster" => CLUSTER      case _ => printErrorAndExit("Deploy mode must be either client or cluster"); -1    }    // Because "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.    ///因为yarn-cluster和yarn-client封装了master和deployMode, 这里对yarn-cluster和yarn-client两种集群管理器和部署模式的组合进行了特殊处理    ///只要知道一个就可以推倒出另一个we have some logic to infer the master and deploy mode from each other    //如果args.master以yarn开头(导致clusterManager == YARN为true)    if (clusterManager == YARN) { ///如果clusterManager是YARN，那么说明master是以yarn开头      if (args.master == "yarn-standalone") { //如果master是yarn-standalone,则提示说master的值yarn-standalone已经不推荐使用，        printWarning("\"yarn-standalone\" is deprecated. Use \"yarn-cluster\" instead.")        args.master = "yarn-cluster" ///将master值改为yarn-cluster      }      (args.master, args.deployMode) match { ///对args.master和args.deployMode的组合进行检查和整理        case ("yarn-cluster", null) => ///如果args.deployMode为null，那么在前面的逻辑中，deployMode的值是CLIENT，          deployMode = CLUSTER //根据args.master的取值yarn-cluster，将deployMode的值改为CLUSTER        case ("yarn-cluster", "client") =>  ////如果args.master, args.deployMode是yarn-cluster，client，那么认为这两个值是冲突的，退出          printErrorAndExit("Client deploy mode is not compatible with master \"yarn-cluster\"")        case ("yarn-client", "cluster") =>  ////如果args.master, args.deployMode是yarn-client，cluster，那么认为这两个值是冲突的，退出          printErrorAndExit("Cluster deploy mode is not compatible with master \"yarn-client\"")        case (_, mode) =>           ///其他情况下，将args.master改为如下值(代码执行到这里的前提是args.master以yarn开头)          ///通过前面的代码，mode只有三种可能的取值：null，cluster和client          //如果mode是client，那么args.master是yarn-client          //如果mode是null，那么args.master是yarn-client          //如果mode是cluster，args.master只能是yarn-cluster，因为(yarn-client，cluster)情况在前面判断过了          args.master = "yarn-" + Option(mode).getOrElse("client")      }      // Make sure YARN is included in our build if we‘re trying to use it      //在YARN的集群模式下检查Spark安装包的完整性，为什么一定要检查org.apache.spark.deploy.yarn.Client类的存在？因为后面的代码中会通过反射调用Client的main方法      //注意，Client是的包名中有个yarn，说明这个Client用于跟YARN集群模式有关      if (!Utils.classIsLoadable("org.apache.spark.deploy.yarn.Client") && !Utils.isTesting) {        printErrorAndExit(          "Could not load YARN classes. " +          "This copy of Spark may not have been compiled with YARN support.")      }    }    // The following modes are not supported or applicable    ///对clusterManager和deployMode不支持的组合进行检查    (clusterManager, deployMode) match {      case (MESOS, CLUSTER) =>///MESOS和CLUSTER不共存        printErrorAndExit("Cluster deploy mode is currently not supported for Mesos clusters.")      case (_, CLUSTER) if args.isPython =>///Python应用程序不支持集群部署模式        printErrorAndExit("Cluster deploy mode is currently not supported for python applications.")      case (_, CLUSTER) if isShell(args.primaryResource) => ///Spark Shell不支持集群部署方式？通过Spark Shell提交Application是一种Client方式？？？        printErrorAndExit("Cluster deploy mode is not applicable to Spark shells.")      case _ =>    }    // If we‘re running a python app, set the main class to our specific python runner    if (args.isPython) { ///如果python应用程序，我表示不关心，忽略之      if (args.primaryResource == PYSPARK_SHELL) {        args.mainClass = "py4j.GatewayServer"        args.childArgs = ArrayBuffer("--die-on-broken-pipe", "0")      } else {        // If a python file is provided, add it to the child arguments and list of files to deploy.        // Usage: PythonAppRunner <main python file> <extra python files> [app arguments]        args.mainClass = "org.apache.spark.deploy.PythonRunner"        args.childArgs = ArrayBuffer(args.primaryResource, args.pyFiles) ++ args.childArgs        args.files = mergeFileLists(args.files, args.primaryResource)      }      args.files = mergeFileLists(args.files, args.pyFiles)      if (args.pyFiles != null) {        sysProps("spark.submit.pyFiles") = args.pyFiles      }    }    // Special flag to avoid deprecation warnings at the client    sysProps("SPARK_SUBMIT") = "true" ///SPARK_SUBMIT这个参数用来做什么的？注意sysProps是个Scala的HashMap    // A list of rules to map each argument to system properties or command-line options in    // each deploy mode; we iterate through these below    ///这是在做什么？？一系列规则（用于在各种部署模式下，把每个argument映射为system properties或者command-line options的规则）    val options = List[OptionAssigner](      // All cluster managers      ///sysProp是指定参数的函数参数传值，此处没有给clOption赋值      OptionAssigner(args.master, ALL_CLUSTER_MGRS, ALL_DEPLOY_MODES, sysProp = "spark.master"),      OptionAssigner(args.name, ALL_CLUSTER_MGRS, ALL_DEPLOY_MODES, sysProp = "spark.app.name"),      ///args.jars指定了提交的Application的jars相关的东西      OptionAssigner(args.jars, ALL_CLUSTER_MGRS, CLIENT, sysProp = "spark.jars"),      OptionAssigner(args.driverMemory, ALL_CLUSTER_MGRS, CLIENT,        sysProp = "spark.driver.memory"),      OptionAssigner(args.driverExtraClassPath, ALL_CLUSTER_MGRS, ALL_DEPLOY_MODES,        sysProp = "spark.driver.extraClassPath"),      OptionAssigner(args.driverExtraJavaOptions, ALL_CLUSTER_MGRS, ALL_DEPLOY_MODES,        sysProp = "spark.driver.extraJavaOptions"),      OptionAssigner(args.driverExtraLibraryPath, ALL_CLUSTER_MGRS, ALL_DEPLOY_MODES,        sysProp = "spark.driver.extraLibraryPath"),      // Standalone cluster only      ///只适用于Standalone集群，属性无需以spark开头，因为都在spark中      OptionAssigner(args.jars, STANDALONE, CLUSTER, sysProp = "spark.jars"),      OptionAssigner(args.driverMemory, STANDALONE, CLUSTER, clOption = "--memory"), //命令行赋值      OptionAssigner(args.driverCores, STANDALONE, CLUSTER, clOption = "--cores"),      // Yarn client only      //只适用于Yarn client，因为由Yarn进行管理，属性名都带有spark      OptionAssigner(args.queue, YARN, CLIENT, sysProp = "spark.yarn.queue"),      OptionAssigner(args.numExecutors, YARN, CLIENT, sysProp = "spark.executor.instances"),      OptionAssigner(args.executorCores, YARN, CLIENT, sysProp = "spark.executor.cores"),      OptionAssigner(args.files, YARN, CLIENT, sysProp = "spark.yarn.dist.files"),      OptionAssigner(args.archives, YARN, CLIENT, sysProp = "spark.yarn.dist.archives"),      // Yarn cluster only      ///只适用于Yarn cluster，为命令行选项赋值      OptionAssigner(args.name, YARN, CLUSTER, clOption = "--name"),      OptionAssigner(args.driverMemory, YARN, CLUSTER, clOption = "--driver-memory"),      OptionAssigner(args.queue, YARN, CLUSTER, clOption = "--queue"),      OptionAssigner(args.numExecutors, YARN, CLUSTER, clOption = "--num-executors"),      OptionAssigner(args.executorMemory, YARN, CLUSTER, clOption = "--executor-memory"),      OptionAssigner(args.executorCores, YARN, CLUSTER, clOption = "--executor-cores"),      OptionAssigner(args.files, YARN, CLUSTER, clOption = "--files"),      OptionAssigner(args.archives, YARN, CLUSTER, clOption = "--archives"),      OptionAssigner(args.jars, YARN, CLUSTER, clOption = "--addJars"),      // Other options      OptionAssigner(args.executorMemory, STANDALONE | MESOS | YARN, ALL_DEPLOY_MODES,        sysProp = "spark.executor.memory"),      OptionAssigner(args.totalExecutorCores, STANDALONE | MESOS, ALL_DEPLOY_MODES,        sysProp = "spark.cores.max"),      OptionAssigner(args.files, LOCAL | STANDALONE | MESOS, ALL_DEPLOY_MODES,        sysProp = "spark.files")    )    // 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 (isUserJar(args.primaryResource)) {        childClasspath += args.primaryResource      }      if (args.jars != null) { childClasspath ++= args.jars.split(",") }      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.sysProp != null) { sysProps.put(opt.sysProp, opt.value) }      }    }    // Add the application jar automatically so the user doesn‘t have to call sc.addJar    // For YARN cluster mode, the jar is already distributed on each node as "app.jar"    // For python files, the primary resource is already distributed as a regular file    val isYarnCluster = clusterManager == YARN && deployMode == CLUSTER    if (!isYarnCluster && !args.isPython) {      var jars = sysProps.get("spark.jars").map(x => x.split(",").toSeq).getOrElse(Seq.empty)      if (isUserJar(args.primaryResource)) {        jars = jars ++ Seq(args.primaryResource)      }      sysProps.put("spark.jars", jars.mkString(","))    }    // In standalone-cluster mode, use Client as a wrapper around the user class    if (clusterManager == STANDALONE && deployMode == CLUSTER) {      childMainClass = "org.apache.spark.deploy.Client"      if (args.supervise) {        childArgs += "--supervise"      }      childArgs += "launch"      childArgs += (args.master, args.primaryResource, args.mainClass)      if (args.childArgs != null) {        childArgs ++= args.childArgs      }    }    // In yarn-cluster mode, use yarn.Client as a wrapper around the user class    if (isYarnCluster) {      childMainClass = "org.apache.spark.deploy.yarn.Client"      if (args.primaryResource != SPARK_INTERNAL) {        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) {      sysProps.getOrElseUpdate(k, v)    }    // Ignore invalid spark.driver.host in cluster modes.    if (deployMode == CLUSTER) {      sysProps -= ("spark.driver.host")    }    // Resolve paths in certain spark properties    val pathConfigs = Seq(      "spark.jars",      "spark.files",      "spark.yarn.jar",      "spark.yarn.dist.files",      "spark.yarn.dist.archives")    pathConfigs.foreach { config =>      // Replace old URIs with resolved URIs, if they exist      sysProps.get(config).foreach { oldValue =http://www.mamicode.com/>        sysProps(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.    sysProps.get("spark.submit.pyFiles").foreach { pyFiles =>      val resolvedPyFiles = Utils.resolveURIs(pyFiles)      val formattedPyFiles = PythonRunner.formatPaths(resolvedPyFiles).mkString(",")      sysProps("spark.submit.pyFiles") = formattedPyFiles    }    (childArgs, childClasspath, sysProps, childMainClass)  }