Standalone cluster模式下的RestSubmissionClientApp

先看看代码,这个类代码比较短,目录是deploy/rest/下。app

private[spark] class RestSubmissionClientApp extends SparkApplication {
  /** Submits a request to run the application and return the response. Visible for testing. */
  def run(
      appResource: String,
      mainClass: String,
      appArgs: Array[String],
      conf: SparkConf,
      env: Map[String, String] = Map()): SubmitRestProtocolResponse = {
    val master = conf.getOption("spark.master").getOrElse {
      throw new IllegalArgumentException("'spark.master' must be set.")
    }
    val sparkProperties = conf.getAll.toMap
    val client = new RestSubmissionClient(master)
    val submitRequest = client.constructSubmitRequest(
      appResource, mainClass, appArgs, sparkProperties, env)
    client.createSubmission(submitRequest)
  }

  override def start(args: Array[String], conf: SparkConf): Unit = {
    if (args.length < 2) {
      sys.error("Usage: RestSubmissionClient [app resource] [main class] [app args*]")
      sys.exit(1)
    }
    val appResource = args(0)
    val mainClass = args(1)
    val appArgs = args.slice(2, args.length)
    val env = RestSubmissionClient.filterSystemEnvironment(sys.env)
    run(appResource, mainClass, appArgs, conf, env)
  }
}

建立一个RestSubmissionClient的client,而后将消息提交给client,消息的格式为:ide

( appResource, mainClass, appArgs, sparkProperties, env)函数

client.createSubmission(submitRequest)ui

client.createSubmission命令作哪些事呢?他就是提交消息给服务端,真实的处理者是服务端,是RestSubmissionServer类或者它的子类。对于独立集群来讲,就是StandaloneRestServer来处理的,咱们就只看submit命令的处理逻辑就能够了。spa

相关的函数有两个:rest

私有方法buildDriverDescription和重写接口方法handleSubmitcode

handleSubmit里调用了前一个方法,最关键的代码是两行:接口

val driverDescription = buildDriverDescription(submitRequest)
val response = masterEndpoint.askSync[DeployMessages.SubmitDriverResponse](
          DeployMessages.RequestSubmitDriver(driverDescription))

生成一个DriverDescription类型的消息,而后给Master发送RequestSubmitDriver消息,让Master来调度执行咱们的spark程序,就是这里的driver。ip

接下来,就进入了Master的处理流程了。get

相关文章
相关标签/搜索