Spark修炼之道（高级篇）——Spark源码阅读：第一节 提交流程

作者：摇摆少年梦   转载自：http://blog.csdn.net/lovehuangjiaju/article/details/49123975

spark-submit 脚本应用程序提交流程

在运行Spar应用程序时，会将Spark应用程序打包后使用spark-submit脚本提交到Spark中运行，执行提交命令如下：

root@sparkmaster:/hadoopLearning/spark-1.5.0-bin-hadoop2.4/bin# 
./spark-submit --master spark://sparkmaster:7077 
--class SparkWordCount --executor-memory 1g
 /root/IdeaProjects/SparkWordCount/out/artifacts/SparkWord
 Count_jar/SparkWordCount.jar  hdfs://ns1/README.md

 hdfs://ns1/SparkWordCountResult

为弄清楚整个流程，我们先来分析一下spark-submit脚本，spark-submit脚本内容如下：

#!/usr/bin/env bash

# disable randomized hash for string in Python 3.3+export PYTHONHASHSEED=0

#spark-submit最终调用的是spark-class脚本#传入的类是org.apache.spark.deploy.SparkSubmit#及其它传入的参数，如deploy mode、executor-memory等exec "$SPARK_HOME"/bin/spark-class org.apach

spark-class脚本会加载spark配置的环境变量信息、定位依赖包spark-assembly-1.5.0-hadoop2.4.0.jar文件（以spark1.5.0为例）等，然后再调用org.apache.spark.launcher.Main正式启动Spark应用程序的运行，具体如下：

# Figure out where Spark is installed#定位SAPRK_HOME目录export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"

# The launcher library will print arguments separated by a NULL character, to allow arguments with# characters that would be otherwise interpreted by the shell. Read that in a while loop, populating# an array that will be used to exec the final command.#执行org.apache.spark.launcher.Main作为Spark应用程序的主入口
CMD=()
while IFS= read -d '' -r ARG; do

  CMD+=("$ARG")

d

从上述代码中，可以看到，通过org.apache.spark.launcher.Main类启动org.apache.spark.deploy.SparkSubmit的执行，SparkSubmit部分源码如下：

//SparkSubmit Main方法
def main(args: Array[String]): Unit = {
    //任务提交时设置的参数，见图2
    val appArgs = new SparkSubmitAarguments(args)
    if (appArgs.verbose) {
      // scalastyle:off println
      printStream.println(appArgs)
      // scalastyle:on println
    }
    appArgs.action match {
      //任务提交时，执行submit(appArgs)
      case SparkSubmitAction.SUBMIT => submit(appArgs)

      case SparkSubmitAction.KILL => kill(appArgs)

 
图1 appArgs = new SparkSubmitAarguments(args)参数

进入submit方法：

  /**
   * 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.
   */
  private def submit(args: SparkSubmitArguments): Unit = {
   //运行参数等信息，见图2
    val (childArgs, childClasspath, sysProps, childMainClass) = prepareSubmitEnvironment(args)
    //定义在submit方法中的方法doRunMain()
    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方法
              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方法
        runMain(childArgs, childClasspath, sysProps, childMainClass, args.verbose)
      }
    }

     // In standalone cluster mode, there are two submission gateways:
     //   (1) The traditional Akka 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
        //调用submit方法中的doRunMain方法
        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.

 
图2 任务提交时设置的参数，

从上面的代码可以看到，最终调用的是runMain方法，其源码如下：

/**
   * Run the main method of the child class using the provided launch environment.
   *
   * Note that this main class will not be the one provided by the user if we're
   * running cluster deploy mode or python applications.
   */
  private def runMain(
      childArgs: Seq[String],
      childClasspath: Seq[String],
      sysProps: Map[String, String],
      childMainClass: String,
      verbose: Boolean): Unit = {
    // scalastyle:off println
    if (verbose) {
      printStream.println(s"Main class:\n$childMainClass")
      printStream.println(s"Arguments:\n${childArgs.mkString("\n")}")
      printStream.println(s"System properties:\n${sysProps.mkString("\n")}")
      printStream.println(s"Classpath elements:\n${childClasspath.mkString("\n")}")
      printStream.println("\n")
    }
    // scalastyle:on println

    try {

      //执行main方法，即执行SparkWordCount

      mainMethod.invoke(null, c

mainMethod.invoke(null, childArgs.toArray)方法执行完毕后，进入SparkWordCount的main方法，执行Spark应用程序，如下图  

至此，正式完成Spark应用程序执行的提交。

打赏

微信扫一扫，打赏作者吧～