Spark 源碼分析（一）：Spark Submit 任務(wù)提交

研究 Spark 內(nèi)部是怎么運(yùn)行的，怎么將 Spark 的任務(wù)從開始運(yùn)行到結(jié)束的踪古，先從 spark-submit 這個 shell 腳本提交用戶程序開始券腔。下面的分析都是基于 spark 2.1.1 版本。

我們一般提交 Spark 任務(wù)時纷纫，都會寫一個如下的腳本，里面指定 spark-submit 腳本的位置辱魁，配置好一些參數(shù)，然后運(yùn)行：

./bin/spark-submit \
  --class <main-class> \
  --master <master-url> \
  --deploy-mode <deploy-mode> \
  --conf <key>=<value> \
  ... # other options
  <application-jar> \
  [application-arguments]

上面那個腳本實(shí)際上會將參數(shù)帶到 spark-submit 腳本中去執(zhí)行参滴，看一下 spark-submit 腳本：

if [ -z "${SPARK_HOME}" ]; then
  source "$(dirname "$0")"/find-spark-home
fi  # disable randomized hash for string in Python 3.3+
export PYTHONHASHSEED=0

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

腳本最后調(diào)用 exec 執(zhí)行 " ${SPARK_HOME}"/bin/spark-class锻弓，調(diào)用的 class 為：`org.apache.spark.deploy.SparkSubmit `，后面的 "\$ @" 是腳本執(zhí)行的所有參數(shù)过蹂。

通過 spark-class 腳本，最終執(zhí)行的命令中酷勺，制定了程序的入口為org.apache.spark.deploy.SparkSubmit。

一脆诉，org.apache.spark.deploy.SparkSubmit

1，main 方法

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

從 main 方法中可以看出亏狰，根據(jù)解析后的參數(shù)中的 action 進(jìn)行模式匹配偶摔，是什么操作就執(zhí)行什么方法，我們這邊是 submit 操作辰斋，則調(diào)用 submit 方法。

2够挂，submit 方法

submit 方法做兩件事情藕夫，第一件事為通過 clusterManager 和 dploymode 去決定下一步要執(zhí)行的類的 main 方法，第二件事是根據(jù)反射執(zhí)行這個 main 方法毅贮。

2.1，submit 方法第一步

這部分主要是準(zhǔn)備下一步要執(zhí)行的相關(guān)類及參數(shù)：

private def submit(args: SparkSubmitArguments): Unit = {
    val (childArgs, childClasspath, sysProps, childMainClass) = prepareSubmitEnvironment(args)

2.1.1病蛉，prepareSubmitEnvironment 方法

通過調(diào)用 prepareSubmitEnvironment 方法來準(zhǔn)備下一步要執(zhí)行的類的 main 方法及相關(guān)參數(shù)，看一下這個方法铡恕，下面這部分是根據(jù)參數(shù)中的 master 和 deploy-mode 來設(shè)置對應(yīng)的 cluasterManager 和部署模式：

private[deploy] def prepareSubmitEnvironment(args: SparkSubmitArguments)
      : (Seq[String], Seq[String], Map[String, String], String) = {
    // 要返回的四個參數(shù)
    val childArgs = new ArrayBuffer[String]()
    val childClasspath = new ArrayBuffer[String]()
    val sysProps = new HashMap[String, String]()
    var childMainClass = ""

    // 根據(jù)腳本中配置的 master 參數(shù)去模式匹配出 clusterManager
    val clusterManager: Int = args.master match {
      case "yarn" => YARN
      case "yarn-client" | "yarn-cluster" =>
        printWarning(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("local") => LOCAL
      case _ =>
        printErrorAndExit("Master must either be yarn or start with spark, mesos, local")
        -1
    }

    // 根據(jù) deployMode 參數(shù)去模式匹配出部署模式
    var deployMode: Int = args.deployMode match {
      case "client" | null => CLIENT
      case "cluster" => CLUSTER
      case _ => printErrorAndExit("Deploy mode must be either client or cluster"); -1
    }

然后會根據(jù)上面匹配出的集群以及部署模式?jīng)Q定怎么提交 application探熔，我們這邊看一下 standalone 集群部署模式，看下面這部分代碼：

        // standalone cluster 模式下的 childMainClass 以及參數(shù)的配置
    if (args.isStandaloneCluster) {
      //如果參數(shù)中配置了 useRest 則為 RestSubmissionClient 的方式去提交 application
      if (args.useRest) {
        childMainClass = "org.apache.spark.deploy.rest.RestSubmissionClient"
        childArgs += (args.primaryResource, args.mainClass)
      } else {
        // 否則使用 Client 放是去提交 application
        childMainClass = "org.apache.spark.deploy.Client"
        if (args.supervise) { childArgs += "--supervise" }
        Option(args.driverMemory).foreach { m => childArgs += ("--memory", m) }
        Option(args.driverCores).foreach { c => childArgs += ("--cores", c) }
        childArgs += "launch"
        childArgs += (args.master, args.primaryResource, args.mainClass)
      }
      if (args.childArgs != null) {
        childArgs ++= args.childArgs
      }
    }

在 standalone 集群模式下诀艰，有兩個提交網(wǎng)關(guān)：

1，使用 org.apache.spark.deploy.Client 作為包裝器來使用傳統(tǒng)的 RPC 網(wǎng)關(guān)苛蒲；

2绿满，使用 Spark 1.3 中引入的基于 rest 的網(wǎng)關(guān)。

2.2喇颁，submit 方法第二步

這里我們的參數(shù)已經(jīng)準(zhǔn)備好了，然后根據(jù)我們 standalone cluster 部署模式?jīng)Q定下一步怎么執(zhí)行：

    /*  在standalone集群模式下橘霎，有兩個提交網(wǎng)關(guān)：
    *   1.使用org.apache.spark.deploy.Client作為包裝器來使用傳統(tǒng)的RPC網(wǎng)關(guān)
    *   2.Spark 1.3中引入的基于rest的網(wǎng)關(guān)
    *   第二種方法是Spark 1.3的默認(rèn)行為，但是Spark submit將會失敗
    *   如果master不是一個REST服務(wù)器瓦盛，那么它將無法使用REST網(wǎng)關(guān)外潜。
    */
        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)
      }
    } else {
      // 其他模式，直接調(diào)用doRunMain方法
      doRunMain()
    }

接著會調(diào)用到 doRunMain 方法橡卤，內(nèi)部其實(shí)調(diào)用了 runMain 方法损搬，所以我們直接看 runMain 方法。

2.2.1嵌灰，runMain 方法

//實(shí)際上這個方法就是根據(jù)我們上面 prepareSubmitEnvironment 方法準(zhǔn)備好的參數(shù)，通過反射的方法去執(zhí)行我們
//下一步要執(zhí)行的類及方法
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

    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)

    for (jar <- childClasspath) {
      addJarToClasspath(jar, loader)
    }

    for ((key, value) <- sysProps) {
      System.setProperty(key, value)
    }

    var mainClass: Class[_] = null

    try {
      mainClass = Utils.classForName(childMainClass)
    } catch {
      case e: ClassNotFoundException =>
        e.printStackTrace(printStream)
        if (childMainClass.contains("thriftserver")) {
          // scalastyle:off println
          printStream.println(s"Failed to load main class $childMainClass.")
          printStream.println("You need to build Spark with -Phive and -Phive-thriftserver.")
          // scalastyle:on println
        }
        System.exit(CLASS_NOT_FOUND_EXIT_STATUS)
      case e: NoClassDefFoundError =>
        e.printStackTrace(printStream)
        if (e.getMessage.contains("org/apache/hadoop/hive")) {
          // scalastyle:off println
          printStream.println(s"Failed to load hive class.")
          printStream.println("You need to build Spark with -Phive and -Phive-thriftserver.")
          // scalastyle:on println
        }
        System.exit(CLASS_NOT_FOUND_EXIT_STATUS)
    }

    // SPARK-4170
    if (classOf[scala.App].isAssignableFrom(mainClass)) {
      printWarning("Subclasses of scala.App may not work correctly. Use a main() method instead.")
    }

    val mainMethod = mainClass.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")
    }

    @tailrec
    def findCause(t: Throwable): Throwable = t match {
      case e: UndeclaredThrowableException =>
        if (e.getCause() != null) findCause(e.getCause()) else e
      case e: InvocationTargetException =>
        if (e.getCause() != null) findCause(e.getCause()) else e
      case e: Throwable =>
        e
    }

    try {
      //通過反射去執(zhí)行準(zhǔn)備好的 mainClass 的 main 方法
      mainMethod.invoke(null, childArgs.toArray)
    } catch {
      case t: Throwable =>
        findCause(t) match {
          case SparkUserAppException(exitCode) =>
            System.exit(exitCode)

          case t: Throwable =>
            throw t
        }
    }
  }

我們選取的 standalone cluster 模式去分析的，根據(jù)上面的 prepareSubmitEnvironment 方法可以知道我們要使用 org.apache.spark.deploy.Client 這個 childMainClass驹溃，然后根據(jù)上面的代碼知道延曙，我們下一步是將相關(guān)參數(shù)帶進(jìn) org.apache.spark.deploy.Client 這個類的 main 方法中去執(zhí)行。

所以下面開始看 org.apache.spark.deploy.Client

二枝缔，org.apache.spark.deploy.Client

Client 用于啟動和終止 standalone 集群中的 Driver 程序。

1，main 方法

def main(args: Array[String]) {
    // scalastyle:off println
    if (!sys.props.contains("SPARK_SUBMIT")) {
      println("WARNING: This client is deprecated and will be removed in a future version of Spark")
      println("Use ./bin/spark-submit with \"--master spark://host:port\"")
    }
    // scalastyle:on println

    val conf = new SparkConf()
    val driverArgs = new ClientArguments(args)

    if (!conf.contains("spark.rpc.askTimeout")) {
      conf.set("spark.rpc.askTimeout", "10s")
    }
    Logger.getRootLogger.setLevel(driverArgs.logLevel)

  //創(chuàng)建 rpcEnv
    val rpcEnv =
      RpcEnv.create("driverClient", Utils.localHostName(), 0, conf, new SecurityManager(conf))

  //獲取 master 節(jié)點(diǎn)的 RpcEndPoint 的引用截型，用于和 master 進(jìn)行 Rpc 通信
    val masterEndpoints = driverArgs.masters.map(RpcAddress.fromSparkURL).
      map(rpcEnv.setupEndpointRef(_, Master.ENDPOINT_NAME))
  
  //注冊 rpcEndpoint儒溉，調(diào)用 onStart方法
    rpcEnv.setupEndpoint("client", new ClientEndpoint(rpcEnv, driverArgs, masterEndpoints, conf))

  //
    rpcEnv.awaitTermination()
  }

這里開始創(chuàng)建 rpcEnv 了，關(guān)于 Rpc 這塊的知識點(diǎn)顿涣，可以看前面這篇文章了解一下：Spark 中的 RPC，拿到 master 的 rpcEndpoint 的引用去注冊 rpcEndpoint舔痪，這里會去調(diào)用 ClientEndpoint 的 onstart 方法。

三锄码，org.apache.spark.deploy.ClientEndpoint

ClientEndPoint 是一個 ThreadSafeRpcEndpoint晌涕，下面看下它的 onStart 方法。

1余黎，onStart 方法

override def onStart(): Unit = {
    driverArgs.cmd match {
      case "launch" =>
        // TODO: We could add an env variable here and intercept it in `sc.addJar` that would
        //       truncate filesystem paths similar to what YARN does. For now, we just require
        //       people call `addJar` assuming the jar is in the same directory.
        val mainClass = "org.apache.spark.deploy.worker.DriverWrapper"

        val classPathConf = "spark.driver.extraClassPath"
        val classPathEntries = sys.props.get(classPathConf).toSeq.flatMap { cp =>
          cp.split(java.io.File.pathSeparator)
        }

        val libraryPathConf = "spark.driver.extraLibraryPath"
        val libraryPathEntries = sys.props.get(libraryPathConf).toSeq.flatMap { cp =>
          cp.split(java.io.File.pathSeparator)
        }

        val extraJavaOptsConf = "spark.driver.extraJavaOptions"
        val extraJavaOpts = sys.props.get(extraJavaOptsConf)
          .map(Utils.splitCommandString).getOrElse(Seq.empty)
        val sparkJavaOpts = Utils.sparkJavaOpts(conf)
        val javaOpts = sparkJavaOpts ++ extraJavaOpts
      
        // 將classPathEntries,libraryPathEntries,javaOpts,drvierArgs信息封裝成Command
        //  這里的mainClass為org.apache.spark.deploy.worker.DriverWrapper
        val command = new Command(mainClass,
          Seq("{{WORKER_URL}}", "{{USER_JAR}}", driverArgs.mainClass) ++ driverArgs.driverOptions,
          sys.env, classPathEntries, libraryPathEntries, javaOpts)
                
        //  將drvierArgs惧财，command信息封裝成DriverDescription
        val driverDescription = new DriverDescription(
          driverArgs.jarUrl,
          driverArgs.memory,
          driverArgs.cores,
          driverArgs.supervise,
          command)
      
        //  向master發(fā)送RequestSubmitDriver,注冊Driver
        ayncSendToMasterAndForwardReply[SubmitDriverResponse](
          RequestSubmitDriver(driverDescription))

      case "kill" =>
        val driverId = driverArgs.driverId
        ayncSendToMasterAndForwardReply[KillDriverResponse](RequestKillDriver(driverId))
    }
  }

這里也會根據(jù) cmd 進(jìn)行模式匹配，垮衷，如果命令為 launch，就去獲取 driver 的額外的 java 依賴搀突，classpath，java 配置甸昏。然后將這些信息封裝為一個 Command 對象，再降 driver 的參數(shù)和 command 信息一起封裝成 DriverDescription 對象施蜜，調(diào)用 ayncSendToMasterAndForwardReply 發(fā)送信息绊寻。

2花墩，ayncSendToMasterAndForwardReply 方法

private def ayncSendToMasterAndForwardReply[T: ClassTag](message: Any): Unit = {
    for (masterEndpoint <- masterEndpoints) {
      masterEndpoint.ask[T](message).onComplete {
        case Success(v) => self.send(v)
        case Failure(e) =>
          logWarning(s"Error sending messages to master $masterEndpoint", e)
      }(forwardMessageExecutionContext)
    }
  }

這個方法實(shí)際上就是將信息發(fā)送到 masterEndpoint 上去。

四和泌，總結(jié)

至此祠肥，我們整個 spark-submit 任務(wù)提交就完成了，接下來就是等待 master 返回 driver 的注冊結(jié)果仇箱，啟動 driver。

最后可以看一下 spark-submit 過程的流程圖：

image