Spark源码阅读1——spark-submit应用提交整体流程

1. spark-submit脚本

  1. 判断${SPARK_HOME}环境变量是否存在，不存在的话调用当前目录下的find-spark-home脚本设置该环境变量。
  2. 执行exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@"命令，调用spark-class脚本。

2. spark-class脚本

  1. 判断${SPARK_HOME}环境变量是否存在，不存在的话调用当前目录下的find-spark-home脚本设置该环境变量。
  2. 执行load-spark-env.sh脚本，设置一些环境变量。
  3. 检查java环境变量。
  4. 检查${SPARK_HOME}中的jars。
  5. 定义如下函数，解析spark-submit脚本传递的参数，并打印解析后的参数。函数中$RUNNER变量是java
1
2
3
4
build_command() {
  "$RUNNER" -Xmx128m -cp "$LAUNCH_CLASSPATH" org.apache.spark.launcher.Main "$@"
  printf "%d\0" $?
}
  1. 执行如下命令，将build_command函数解析后返回的参数赋值给CMD。
1
2
3
4
5
set +o posix
CMD=()
while IFS= read -d '' -r ARG; do
  CMD+=("$ARG")
done < <(build_command "$@")
  1. 执行CMD中的摸个参数类，若是提交应用程序，则参数类为：org.apache.spark.deploy.SparkSubmit。SparkSubmit参数类会完成启动Spark应用程序的工作。
1
2
CMD=("${CMD[@]:0:$LAST}")
exec "${CMD[@]}"

3. org.apache.spark.launcher.Main类

该类为Spark启动器的命令行接口，被Spark脚本内部使用，主要实现对传入参数的解析工作。最终打印出解析后的参数格式，类似于java -cp 各种option 各种路径 org.apache.spark.deploy.SparkSubmit --master .. --deploy-mode .. --class 用户主类 其他参数

3.1 main方法

  1. main方法中将传入的第一个参数解析为className，若className为org.apache.spark.deploy.SparkSubmit类，则构建 builder = new SparkSubmitCommandBuilder(args)；否则构建builder = new SparkClassCommandBUilder(className, args)。
  2. 构建builder后，执行List<String> cmd = builder.buildCommand(env)方法，获取解析后的命令行cmd。
  3. 将cmd根据不同的运行环境(Windows/Linux)打印出来。spark-class脚本会根据$?获取到main方法最后打印的数据。
1
2
3
4
5
6
7
8
9
10
if (isWindows()) {
  System.out.println(prepareWindowsCommand(cmd, env));
} else {
  // In bash, use NULL as the arg separator since it cannot be used in an argument.
  List<String> bashCmd = prepareBashCommand(cmd, env);
  for (String c : bashCmd) {
    System.out.print(c);
    System.out.print('\0');
  }
}

3.2 org.apache.spark.launcher.SparkSubmitCommandBuilder类

  1. 该类的构造器中会调用OptionParser类的parse(submitArgs)方法，将传入的参数通过正则表达式Pattern eqSeparatedOpt = Pattern.compile("(--[^=]+)=(.+)")匹配后，调用OptionParser的handle方法，这个即可解析出master、deployMode、propertiesFile、driveMermory等参数。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
protected boolean handle(String opt, String value) {
  if (opt.equals(MASTER)) {
    master = value;
  } else if (opt.equals(DEPLOY_MODE)) {
    deployMode = value;
  } else if (opt.equals(PROPERTIES_FILE)) {
    propertiesFile = value;
  } else if (opt.equals(DRIVER_MEMORY)) {
    conf.put(SparkLauncher.DRIVER_MEMORY, value);
  } else if (opt.equals(DRIVER_JAVA_OPTIONS)) {
    conf.put(SparkLauncher.DRIVER_EXTRA_JAVA_OPTIONS, value);
  } else if (opt.equals(DRIVER_LIBRARY_PATH)) {
    conf.put(SparkLauncher.DRIVER_EXTRA_LIBRARY_PATH, value);
  } else if (opt.equals(DRIVER_CLASS_PATH)) {
    conf.put(SparkLauncher.DRIVER_EXTRA_CLASSPATH, value);
  } else if (opt.equals(CONF)) {
    String[] setConf = value.split("=", 2);
    checkArgument(setConf.length == 2, "Invalid argument to %s: %s", CONF, value);
    conf.put(setConf[0], setConf[1]);
  } else if (opt.equals(CLASS)) {
    // The special classes require some special command line handling, since they allow
    // mixing spark-submit arguments with arguments that should be propagated to the shell
    // itself. Note that for this to work, the "--class" argument must come before any
    // non-spark-submit arguments.
    mainClass = value;
    if (specialClasses.containsKey(value)) {
      allowsMixedArguments = true;
      appResource = specialClasses.get(value);
    }
  } else if (opt.equals(KILL_SUBMISSION) || opt.equals(STATUS)) {
    isAppResourceReq = false;
    sparkArgs.add(opt);
    sparkArgs.add(value);
  } else if (opt.equals(HELP) || opt.equals(USAGE_ERROR)) {
    isAppResourceReq = false;
    sparkArgs.add(opt);
  } else if (opt.equals(VERSION)) {
    isAppResourceReq = false;
    sparkArgs.add(opt);
  } else {
    sparkArgs.add(opt);
    if (value != null) {
      sparkArgs.add(value);
    }
  }
  return true;
}
  1. buildSparkSubmitCommand(Map<String, String> env)方法。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
private List<String> buildSparkSubmitCommand(Map<String, String> env)
    throws IOException, IllegalArgumentException {
  // Load the properties file and check whether spark-submit will be running the app's driver
  // or just launching a cluster app. When running the driver, the JVM's argument will be
  // modified to cover the driver's configuration.
  Map<String, String> config = getEffectiveConfig();
  boolean isClientMode = isClientMode(config);
  String extraClassPath = isClientMode ? config.get(SparkLauncher.DRIVER_EXTRA_CLASSPATH) : null;

  List<String> cmd = buildJavaCommand(extraClassPath);
  // Take Thrift Server as daemon
  if (isThriftServer(mainClass)) {
    addOptionString(cmd, System.getenv("SPARK_DAEMON_JAVA_OPTS"));
  }
  addOptionString(cmd, System.getenv("SPARK_SUBMIT_OPTS"));
  addOptionString(cmd, System.getenv("SPARK_JAVA_OPTS"));

  // We don't want the client to specify Xmx. These have to be set by their corresponding
  // memory flag --driver-memory or configuration entry spark.driver.memory
  String driverExtraJavaOptions = config.get(SparkLauncher.DRIVER_EXTRA_JAVA_OPTIONS);
  if (!isEmpty(driverExtraJavaOptions) && driverExtraJavaOptions.contains("Xmx")) {
    String msg = String.format("Not allowed to specify max heap(Xmx) memory settings through " +
                 "java options (was %s). Use the corresponding --driver-memory or " +
                 "spark.driver.memory configuration instead.", driverExtraJavaOptions);
    throw new IllegalArgumentException(msg);
  }

  if (isClientMode) {
    // Figuring out where the memory value come from is a little tricky due to precedence.
    // Precedence is observed in the following order:
    // - explicit configuration (setConf()), which also covers --driver-memory cli argument.
    // - properties file.
    // - SPARK_DRIVER_MEMORY env variable
    // - SPARK_MEM env variable
    // - default value (1g)
    // Take Thrift Server as daemon
    String tsMemory =
      isThriftServer(mainClass) ? System.getenv("SPARK_DAEMON_MEMORY") : null;
    String memory = firstNonEmpty(tsMemory, config.get(SparkLauncher.DRIVER_MEMORY),
      System.getenv("SPARK_DRIVER_MEMORY"), System.getenv("SPARK_MEM"), DEFAULT_MEM);
    cmd.add("-Xmx" + memory);
    addOptionString(cmd, driverExtraJavaOptions);
    mergeEnvPathList(env, getLibPathEnvName(),
      config.get(SparkLauncher.DRIVER_EXTRA_LIBRARY_PATH));
  }

  addPermGenSizeOpt(cmd);
  cmd.add("org.apache.spark.deploy.SparkSubmit");
  cmd.addAll(buildSparkSubmitArgs());
  return cmd;
}
  • 该方法首先根据propertiesFile参数加载配置文件中的对应配置项，若propertiesFile为null，则从默认的配置文件spark-defaults.conf中加载。其中spark-defaults.conf从SPARK_CONF_DIR或SPARK_HOME/conf中获取。
  • 调用buildJavaCommand(extraClassPath)方法，该方法会构建一个运行java程序的参数列表。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
List<String> buildJavaCommand(String extraClassPath) throws IOException {
  List<String> cmd = new ArrayList<>();
  String envJavaHome;

  if (javaHome != null) {
    cmd.add(join(File.separator, javaHome, "bin", "java"));
  } else if ((envJavaHome = System.getenv("JAVA_HOME")) != null) {
      cmd.add(join(File.separator, envJavaHome, "bin", "java"));
  } else {
      cmd.add(join(File.separator, System.getProperty("java.home"), "bin", "java"));
  }

  // Load extra JAVA_OPTS from conf/java-opts, if it exists.
  File javaOpts = new File(join(File.separator, getConfDir(), "java-opts"));
  if (javaOpts.isFile()) {
    BufferedReader br = new BufferedReader(new InputStreamReader(
        new FileInputStream(javaOpts), StandardCharsets.UTF_8));
    try {
      String line;
      while ((line = br.readLine()) != null) {
        addOptionString(cmd, line);
      }
    } finally {
      br.close();
    }
  }

  cmd.add("-cp");
  cmd.add(join(File.pathSeparator, buildClassPath(extraClassPath)));
  return cmd;
}
  • List<String> cmd构建好以后，cmd中已经有了java -cp等运行java的命令参数，后面添加-Xmx memory、spark.driver.extraJavaOptions、spark.driver.extraLibraryPath等额外参数，最后添加org.apache.spark.deploy.SparkSubmit和调用buildSparkSubmitArgs()方法，添加SparkSubmit相应的参数，包括–master、–deployMode、–conf、–class和用户程序对应的传入参数。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
List<String> buildSparkSubmitArgs() {
  List<String> args = new ArrayList<>();
  SparkSubmitOptionParser parser = new SparkSubmitOptionParser();

  if (!allowsMixedArguments && isAppResourceReq) {
    checkArgument(appResource != null, "Missing application resource.");
  }

  if (verbose) {
    args.add(parser.VERBOSE);
  }

  if (master != null) {
    args.add(parser.MASTER);
    args.add(master);
  }

  if (deployMode != null) {
    args.add(parser.DEPLOY_MODE);
    args.add(deployMode);
  }

  if (appName != null) {
    args.add(parser.NAME);
    args.add(appName);
  }

  for (Map.Entry<String, String> e : conf.entrySet()) {
    args.add(parser.CONF);
    args.add(String.format("%s=%s", e.getKey(), e.getValue()));
  }

  if (propertiesFile != null) {
    args.add(parser.PROPERTIES_FILE);
    args.add(propertiesFile);
  }

  if (isExample) {
    jars.addAll(findExamplesJars());
  }

  if (!jars.isEmpty()) {
    args.add(parser.JARS);
    args.add(join(",", jars));
  }

  if (!files.isEmpty()) {
    args.add(parser.FILES);
    args.add(join(",", files));
  }

  if (!pyFiles.isEmpty()) {
    args.add(parser.PY_FILES);
    args.add(join(",", pyFiles));
  }

  if (isAppResourceReq) {
    checkArgument(!isExample || mainClass != null, "Missing example class name.");
  }
  if (mainClass != null) {
    args.add(parser.CLASS);
    args.add(mainClass);
  }

  args.addAll(sparkArgs);
  if (appResource != null) {
    args.add(appResource);
  }
  args.addAll(appArgs);

  return args;
}

其中sparkArgs是SparkSubmitCommandBuilder类未解析的spark相应配置，如–executor-memory等参数。appArgs为用户主类需要用到的参数。

4. org.apache.spark.deploy.SparkSubmit类

spark-class脚本中最后会执行exec java -cp ... org.apache.spark.deploy.SparkSubmit ...来运行SparkSubmit类。该类是启动Spark Application的主入口。

4.1 main方法

该方法首先会根据传入的args创建SparkSubmitArguments对象appArgs，然后根据appArgs.action的不同值调用相应的submit(appArgs)、kill(appArgs)、requestStatus(appArgs)方法，后两种方法只standalone和Mesos cluster模式下支持。

1
2
3
4
5
6
7
8
9
10
11
12
13
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)
  }
}

4.2 SparkSubmitArguments类

该类包含了运行Spark Application所需要的参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
var master: String = null
var deployMode: String = null
var executorMemory: String = null
var executorCores: String = null
var totalExecutorCores: String = null
var propertiesFile: String = null
var driverMemory: String = null
var driverExtraClassPath: String = null
var driverExtraLibraryPath: String = null
var driverExtraJavaOptions: String = null
var queue: String = null
var numExecutors: String = null
var files: String = null
var archives: String = null
var mainClass: String = null
var primaryResource: String = null
var name: String = null
var childArgs: ArrayBuffer[String] = new ArrayBuffer[String]()
var jars: String = null
var packages: String = null
var repositories: String = null
var ivyRepoPath: String = null
var packagesExclusions: String = null
var verbose: Boolean = false
var isPython: Boolean = false
var pyFiles: String = null
var isR: Boolean = false
var action: SparkSubmitAction = null
val sparkProperties: HashMap[String, String] = new HashMap[String, String]()
var proxyUser: String = null
var principal: String = null
var keytab: String = null

SparkSubmitArguments类的构造器中，一次调用了下面五种方法，最终将所有需要使用到的参数都解析到了SparkSubmitArguments类相应的字段中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// Set parameters from command line arguments
try {
  parse(args.asJava)
} catch {
  case e: IllegalArgumentException =>
    SparkSubmit.printErrorAndExit(e.getMessage())
}
// Populate `sparkProperties` map from properties file
mergeDefaultSparkProperties()
// Remove keys that don't start with "spark." from `sparkProperties`.
ignoreNonSparkProperties()
// Use `sparkProperties` map along with env vars to fill in any missing parameters
loadEnvironmentArguments()

validateArguments()

4.2.1 parse(args.asJava)方法

SparkSubmitOptionParser类的parse(submitArgs)方法，将传入的参数通过正则表达式Pattern eqSeparatedOpt = Pattern.compile("(--[^=]+)=(.+)")匹配后，调用SparkSubmitArguments的handle方法，使用Scala的模式匹配从submitArgs中解析出相应参数并赋值给SparkSubmitArguments的相应字段，其中用户主类对应的参数存放在SparkSubmitArguments的childArgs字段中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
/** Fill in values by parsing user options. */
override protected def handle(opt: String, value: String): Boolean = {
  opt match {
    case NAME =>
      name = value

    case MASTER =>
      master = value

    case CLASS =>
      mainClass = value

    case DEPLOY_MODE =>
      if (value != "client" && value != "cluster") {
        SparkSubmit.printErrorAndExit("--deploy-mode must be either \"client\" or \"cluster\"")
      }
      deployMode = value

    case NUM_EXECUTORS =>
      numExecutors = value

    case TOTAL_EXECUTOR_CORES =>
      totalExecutorCores = value

    case EXECUTOR_CORES =>
      executorCores = value

    case EXECUTOR_MEMORY =>
      executorMemory = value
    ......
    ......
  }
    true
}

4.2.2 mergeDefaultSparkProperties()方法

sparkProperties已经在前面的handle方法中将用户--conf命令提交的相关配置加载过，mergeDefaultSparkProperties()主要完成加载propertiesFile配置文件中的相关配置，若propertiesFile没有配置则使用默认的spark-default.conf。需要注意的是sparkProperties已经在—conf配置过的数据不会被覆盖，即--conf比配置文件的优先级高。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
 * Merge values from the default properties file with those specified through --conf.
 * When this is called, `sparkProperties` is already filled with configs from the latter.
 */
private def mergeDefaultSparkProperties(): Unit = {
  // Use common defaults file, if not specified by user
  propertiesFile = Option(propertiesFile).getOrElse(Utils.getDefaultPropertiesFile(env))
  // Honor --conf before the defaults file
  defaultSparkProperties.foreach { case (k, v) =>
    // sparkProperties之前已经在--conf中配置过的数据不会被覆盖
    if (!sparkProperties.contains(k)) {
      sparkProperties(k) = v
    }
  }
}

4.2.3 ignoreNonSparkProperties()

该方法主要是去除掉sparkProperties中不是spark.开头的无效配置项。

1
2
3
4
5
6
7
8
9
10
11
/**
 * Remove keys that don't start with "spark." from `sparkProperties`.
 */
private def ignoreNonSparkProperties(): Unit = {
  sparkProperties.foreach { case (k, v) =>
    if (!k.startsWith("spark.")) {
      sparkProperties -= k
      SparkSubmit.printWarning(s"Ignoring non-spark config property: $k=$v")
    }
  }
}

4.2.4 loadEnvironmentArguments()

该方法主要对SparkSubmitArguments类中的一些必备的字段进行验证，若该字段未设置，则从sparkProperties中提取，若sparkProperties中不存在，则从环境变量中取，若仍然不存在，则取Null或相应的默认值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
  * Load arguments from environment variables, Spark properties etc.
  */
 private def loadEnvironmentArguments(): Unit = {
   master = Option(master)
     .orElse(sparkProperties.get("spark.master"))
     .orElse(env.get("MASTER"))
     .orNull
   driverExtraClassPath = Option(driverExtraClassPath)
     .orElse(sparkProperties.get("spark.driver.extraClassPath"))
     .orNull
   driverExtraJavaOptions = Option(driverExtraJavaOptions)
     .orElse(sparkProperties.get("spark.driver.extraJavaOptions"))
     .orNull
   ......
   ......
 }

4.2.5 validateArguments()方法

该方法用来确保所有要求的字段存在。

1
2
3
4
5
6
7
8
/** Ensure that required fields exists. Call this only once all defaults are loaded. */
private def validateArguments(): Unit = {
  action match {
    case SUBMIT => validateSubmitArguments()
    case KILL => validateKillArguments()
    case REQUEST_STATUS => validateStatusRequestArguments()
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
private def validateSubmitArguments(): Unit = {
   if (args.length == 0) {
     printUsageAndExit(-1)
   }
   if (primaryResource == null) {
     SparkSubmit.printErrorAndExit("Must specify a primary resource (JAR or Python or R file)")
   }
   if (mainClass == null && SparkSubmit.isUserJar(primaryResource)) {
     SparkSubmit.printErrorAndExit("No main class set in JAR; please specify one with --class")
   }
   if (pyFiles != null && !isPython) {
     SparkSubmit.printErrorAndExit("--py-files given but primary resource is not a Python script")
   }

   if (master.startsWith("yarn")) {
     val hasHadoopEnv = env.contains("HADOOP_CONF_DIR") || env.contains("YARN_CONF_DIR")
     if (!hasHadoopEnv && !Utils.isTesting) {
       throw new Exception(s"When running with master '$master' " +
         "either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment.")
     }
   }

   if (proxyUser != null && principal != null) {
     SparkSubmit.printErrorAndExit("Only one of --proxy-user or --principal can be provided.")
   }
 }

4.3 submit(args:SparkSubmitArguments)方法

在将传入的args字符串数组构造成SparkSubmitArguments对象appArgs后调用submit(appArgs)方法。该方法根据提供的参数，提交Spark Application。主要分为两步：第一步是为main class准备启动环境，包含设置相应的classpath、sysytem properties、application arguments等，第二步是使用反射机制运行main class的main方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
/**
 * Submit the application using the provided parameters.
 *
 * 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): Unit = {
  val (childArgs, childClasspath, sysProps, 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, sysProps, 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) {
            // scalastyle:off println
            printStream.println(s"ERROR: ${e.getClass().getName()}: ${e.getMessage()}")
            // scalastyle:on println
            exitFn(1)
          } else {
            throw e
          }
      }
    } else {
      runMain(childArgs, childClasspath, sysProps, childMainClass, args.verbose)
    }
  }

   // 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 {
      // scalastyle:off println
      printStream.println("Running Spark using the REST application submission protocol.")
      // scalastyle:on println
      doRunMain()
    } catch {
      // Fail over to use the legacy submission gateway
      case e: SubmitRestConnectionException =>
        printWarning(s"Master endpoint ${args.master} was not a REST server. " +
          "Falling back to legacy submission gateway instead.")
        args.useRest = false
        submit(args)
    }
  // In all other modes, just run the main class as prepared
  } else {
    doRunMain()
  }
}

4.3.1 prepareSubmitEnvironment(args)方法

该方法是为提交spark application准备环境，返回4种数据：1. childArgs——子进程参数；2.childClasspath——子进程类路径列表；3. sysProps——sysytem properties map；4. childMainClass——子进程主类。

该方法会根据args判断用户设定的master、deployMode等，根据用户要求的运行模式准备相应的运行环境。其中也会根据不同的master设置不同的childMainClass。比如Client模式下会直接将childMainClass=args.mainClass；YarnCluster模式下，会将childMainClass = "org.apache.spark.deploy.yarn.Client"，将用户的主类通过childArgs += ("--class", args.mainClass)设置。同时也会把用户传入的jar包，添加到chlidClasspath中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
......
......
// 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
    // Also add the main application jar and any added jars to classpath in case YARN client
    // requires these jars.
    if (deployMode == CLIENT || isYarnCluster) {
      childMainClass = args.mainClass
      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 }
    }

    // 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.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) }
      }
    }
......
......

4.3.2 runMain(childArgs, childClasspath, sysProps, childMainClass, verbose)方法

该方法使用反射机制启动childMainClass中的main方法。需要特别注意的是，从上一步也可以了解到，当运行在cluster deploy mode或python application时，childMainClass并不是用户提供的主类。如YarnCluster模式下childMainClass = "org.apache.spark.deploy.yarn.Client"。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// 设置类加载器
val loader =
      if (sysProps.getOrElse("spark.driver.userClassPathFirst", "false").toBoolean) {
        new ChildFirstURLClassLoader(new Array[URL](0),
          Thread.currentThread.getContextClassLoader)
      } else {
        new MutableURLClassLoader(new Array[URL](0),
          Thread.currentThread.getContextClassLoader)
      }
    Thread.currentThread.setContextClassLoader(loader)
	// 根据已经设置的childClasspath，将对应的jar包文件路径配置到类加载器的加载路径中
    for (jar <- childClasspath) {
      addJarToClasspath(jar, loader)
    }
    // 根据已经设置的sysProps，配置系统参数
    for ((key, value) <- sysProps) {
      System.setProperty(key, value)
    }
......
......
// 根据childMainClass获取主类
mainClass = Utils.classForName(childMainClass)
......
......
// 利用反射机制获取mainClass中的主方法mainMethond，并启动主方法。
val mainMethod = mainClass.getMethod("main", new Array[String](0).getClass)
mainMethod.invoke(null, childArgs.toArray)

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    2、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: true
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

前行的路上<br><br>点点滴滴……