Spark2.x精通:从spark-submit提交到driver启动

接上篇文章： Spark2.2.0精通：详解Master端任务调度schedule()函数 ，这里从client出发，详细用户通过spark-submit提交任务后，Spark集群如何进行提交命令处理；由于上篇文章主要只提到了Master端如何进行Executor的启动，没有讲解Driver的启动，这里结合spark-submit任务提交，把Driver的提交和启动一块讲了。

1.spark-submit提交任务，调用的脚本spark-submit.sh，脚本基本没啥操作，直接调用org.apache.spark.deploy.SparkSubmit类，比较简单；

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 "$@"

2.直接找到SparkSubmit的main()函数，代码如下：

override 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 {
      //调用submit()函数，参数是用户提交指定参数
      case SparkSubmitAction.SUBMIT => submit(appArgs)
      case SparkSubmitAction.KILL => kill(appArgs)
      case SparkSubmitAction.REQUEST_STATUS => requestStatus(appArgs)
    }
  }

3.submit()函数，主要是进行了一些初始化，然后提交了Driver

 @tailrec
  private def submit(args: SparkSubmitArguments): Unit = {
  //prepareSubmitEnvironment()函数代码比较多，主要干了以下几件事：
   // 1.根据用户参数，主要是master和deploye-mode，设置任务的提交模式
   // 2.根据提交模式实例化主类，
    //childArgs： 主要就是一些参数的
    //childClasspath：这个就是classPath，jvm运行的class路径
    //sysProps：一些系统参数
    // childMainClass：接下来将要运行的主类
    val (childArgs, childClasspath, sysProps, childMainClass) = prepareSubmitEnvironment(args)

    def doRunMain(): Unit = {
      //  proxyUser 一般为null 直接看下面
      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，主要是打印了一些日志信息
        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()去执行 
      doRunMain()
    }
  }

4.这里才是spark-submit的核心，调用doRunMain()，他里面没啥东西，一般我们不设置代理，直接执行else：

 /*  在standalone集群模式下，有两个提交网关：
    *   1.使用org.apache.spark.deploy.Client作为包装器来使用传统的RPC网关
    *   2.Spark 1.3中引入的基于rest的网关
    *   第二种方法是Spark 1.3的默认行为，但是Spark submit将会失败
    *   如果master不是一个REST服务器，那么它将无法使用REST网关。
    */
  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)
      }
    }

5.最后调用runMain()函数，代码如下：

 //当deploy mode为client时，执行用户自己编写的主方法
// 当deploy mode为cluster时，需要判断是否为REST提交，如果是则执行
// org.apache.spark.rest.RestSubmissionClient的主方法，如果不是则执行
// org.apache.spark.deploy.Client的主方法
  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")}")
      // sysProps may contain sensitive information, so redact before printing
      printStream.println(s"System properties:\n${Utils.redact(sysProps).mkString("\n")}")
      printStream.println(s"Classpath elements:\n${childClasspath.mkString("\n")}")
      printStream.println("\n")
    }
    // scalastyle:on println
    //由于默认情况下，优先级SPARK_HOME/lib/jar包 > 用户程序中的jar包， 
    // 如果想让用户程序jar优先执行，那么要使用 spark.yarn.user.classpath.first (spark1.3以前）或者
    //  spark.executor.userClassPathFirst 和spark.driver.userClassPathFirst 参数。
    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)
    //使用URLClassLoader加载jar包
    for (jar <- childClasspath) {
      addJarToClasspath(jar, loader)
    }

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

    var mainClass: Class[_] = null
    
    try {
    //获取用户指定的Main函数
      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 {
    //mainMethod.invoke 是通过反射来调用的 对应的主函数
      mainMethod.invoke(null, childArgs.toArray)
    } catch {
      case t: Throwable =>
        findCause(t) match {
          case SparkUserAppException(exitCode) =>
            System.exit(exitCode)

          case t: Throwable =>
            throw t
        }
    }
  }

6.由于我们这里是Standalone模式，会映射到 org.apache.spark.deploy.Client 直接调用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)

  //准备rpcEnv环境，之后通过master的地址获取masterEndpoints端点相关信息，
  // 通过rpcEnv注册相关clientEndPoint端点信息，同时需要注意，这里会把masterEndpoints上
  //端点信息也作为构造ClientEndpoint端点的参数，也就是说这个ClientEndpoint会和masterEndpoints通信
    val rpcEnv =
      RpcEnv.create("driverClient", Utils.localHostName(), 0, conf, new SecurityManager(conf))

    val masterEndpoints = driverArgs.masters.map(RpcAddress.fromSparkURL).
      map(rpcEnv.setupEndpointRef(_, Master.ENDPOINT_NAME))
     //之前说过只要调用setupEndpoint就会调用Onstart()函数，由于我们是提交任务
     // 这里会直接调用“launch”事件处理函数
    rpcEnv.setupEndpoint("client", new ClientEndpoint(rpcEnv, driverArgs, masterEndpoints, conf))

    rpcEnv.awaitTermination()
  }

7.我们直接看onStart()函数，代码如下：

case "launch" =>
        //这里会启动DriverWrapper类
        val mainClass = "org.apache.spark.deploy.worker.DriverWrapper"
       
       //下面都是指定了一些classPath libpath 不再讲解
        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
        val command = new Command(mainClass,
          Seq("{{WORKER_URL}}", "{{USER_JAR}}", driverArgs.mainClass) ++ driverArgs.driverOptions,
          sys.env, classPathEntries, libraryPathEntries, javaOpts)
        //DriverWrapper封装到command中，然后封装到driverDescription中
        val driverDescription = new DriverDescription(
          driverArgs.jarUrl,
          driverArgs.memory,
          driverArgs.cores,
          driverArgs.supervise,
          command)
          //这里会调用RequestSubmitDriver，向Master发送RequestSubmitDriver消息
          //注意这里Master接收消息，处理完成后会进行回复的，回复SubmitDriverResponse消息
          //会在receive()函数进行处理
        ayncSendToMasterAndForwardReply[SubmitDriverResponse](
          RequestSubmitDriver(driverDescription))

8.Master端接收到 RequestSubmitDriver 消息，在receiveAndReply()函数进行对应事件的处理，代码如下：

case RequestSubmitDriver(description) =>
      //如果数据恢复状态，直接跳过
      if (state != RecoveryState.ALIVE) {
        val msg = s"${Utils.BACKUP_STANDALONE_MASTER_PREFIX}: $state. " +
          "Can only accept driver submissions in ALIVE state."
        context.reply(SubmitDriverResponse(self, false, None, msg))
      } else {
        //创建driver，然后加入持久化引擎中，将driver添加到等待队列
        logInfo("Driver submitted " + description.command.mainClass)
        val driver = createDriver(description)
        persistenceEngine.addDriver(driver)
        waitingDrivers += driver
        drivers.add(driver)
        //资源调度函数已经在上一篇文章讲解：Spark2.2.0精通：详解Master端任务调度schedule()函数
        schedule()

        // TODO: It might be good to instead have the submission client poll the master to determine
        //       the current status of the driver. For now it's simply "fire and forget".
        context.reply(SubmitDriverResponse(self, true, Some(driver.id),
          s"Driver successfully submitted as ${driver.id}"))
      }

至此，用户通过spark-submit提交任务到Driver启动的整个流程基本都梳理清楚了。