Spark 之SparkContext 源码精读2
Spark 之SparkContext 源码精读2
柯里昂 发表于2年前
Spark 之SparkContext 源码精读2
  • 发表于 2年前
  • 阅读 146
  • 收藏 1
  • 点赞 1
  • 评论 0

前文说到SparkContext的3大核心对象被创建。

分别是TaskScheduler、SchedulerBackend、DAGScheduler。

 

这三大对象创建完成后,紧接着,调用了TaskScheduler的start方法。

// SparkContext.scala line 527

// start TaskScheduler after taskScheduler sets DAGScheduler reference in DAGScheduler's
// constructor
_taskScheduler.start()

上面的注释也很好的说明了,在调用TaskScheduler.start 前,需要在DAGScheduler的构造中,设置TaskScheduler的DAGScheduler的引用。这个在前文创建DAGScheduler的时候已经说明,他们之间是相互引用的。

 

当TaskScheduler.start被调用时,我们看看具体发生了什么。

// TaskSchedulerImpl.scala line 143
override def start() {
  backend.start()

  if (!isLocal && conf.getBoolean("spark.speculation", false)) {
    logInfo("Starting speculative execution thread")
    speculationScheduler.scheduleAtFixedRate(new Runnable {
      override def run(): Unit = Utils.tryOrStopSparkContext(sc) {
        checkSpeculatableTasks()
      }
    }, SPECULATION_INTERVAL_MS, SPECULATION_INTERVAL_MS, TimeUnit.MILLISECONDS)
  }
}

首先:beakend.start被调用了。这里提到,TaskScheduler中有ShedulerBackend的引用。我们也认为SchedulerBackend就是TaskScheduler的Backend。因此在start的时候,Backend要先start。这样应该会容易理解。

那跟踪进SchedulerBackend的start方法中瞧瞧吧。这里的SchedulerBackend是SparkDeploySchedulerBackend的实例。

// SparkDeploySchedulerBackend.scala line 52
override def start() {
  super.start()
  launcherBackend.connect()

  // The endpoint for executors to talk to us
  val driverUrl = rpcEnv.uriOf(SparkEnv.driverActorSystemName,
    RpcAddress(sc.conf.get("spark.driver.host"), sc.conf.get("spark.driver.port").toInt),
    CoarseGrainedSchedulerBackend.ENDPOINT_NAME)
  val args = Seq(
    "--driver-url", driverUrl,
    "--executor-id", "{{EXECUTOR_ID}}",
    "--hostname", "{{HOSTNAME}}",
    "--cores", "{{CORES}}",
    "--app-id", "{{APP_ID}}",
    "--worker-url", "{{WORKER_URL}}")
  val extraJavaOpts = sc.conf.getOption("spark.executor.extraJavaOptions")
    .map(Utils.splitCommandString).getOrElse(Seq.empty)
  val classPathEntries = sc.conf.getOption("spark.executor.extraClassPath")
    .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)
  val libraryPathEntries = sc.conf.getOption("spark.executor.extraLibraryPath")
    .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)

  // When testing, expose the parent class path to the child. This is processed by
  // compute-classpath.{cmd,sh} and makes all needed jars available to child processes
  // when the assembly is built with the "*-provided" profiles enabled.
  val testingClassPath =
    if (sys.props.contains("spark.testing")) {
      sys.props("java.class.path").split(java.io.File.pathSeparator).toSeq
    } else {
      Nil
    }

  // Start executors with a few necessary configs for registering with the scheduler
  val sparkJavaOpts = Utils.sparkJavaOpts(conf, SparkConf.isExecutorStartupConf)
  val javaOpts = sparkJavaOpts ++ extraJavaOpts
  val command = Command("org.apache.spark.executor.CoarseGrainedExecutorBackend",
    args, sc.executorEnvs, classPathEntries ++ testingClassPath, libraryPathEntries, javaOpts)
  val appUIAddress = sc.ui.map(_.appUIAddress).getOrElse("")
  val coresPerExecutor = conf.getOption("spark.executor.cores").map(_.toInt)
  val appDesc = new ApplicationDescription(sc.appName, maxCores, sc.executorMemory,
    command, appUIAddress, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor)
  client = new AppClient(sc.env.rpcEnv, masters, appDesc, this, conf)
  client.start()
  launcherBackend.setState(SparkAppHandle.State.SUBMITTED)
  waitForRegistration()
  launcherBackend.setState(SparkAppHandle.State.RUNNING)
}

在start方法中,先调用了super.start,上文可知,super是CoarseGrainedSchedulerBackend。

在父类的start方法中,除了设置spark.开头的配置外,做了一件很重要的事情:创建了一个Endpoint,为DriverEndpoint。

// CoarseGrainedSchedulerBackend.scala line 303

override def start() {
  val properties = new ArrayBuffer[(String, String)]
  for ((key, value) <- scheduler.sc.conf.getAll) {
    if (key.startsWith("spark.")) {
      properties += ((key, value))
    }
  }

  // TODO (prashant) send conf instead of properties
  driverEndpoint = rpcEnv.setupEndpoint(ENDPOINT_NAME, createDriverEndpoint(properties)) // 创建DriverEndpoint并向RpcEnv注册
}

Endpoint的名字叫:CoarseGrainedScheduler

// CoarseGrainedSchedulerBackend.scala line 510

private[spark] object CoarseGrainedSchedulerBackend {
  val ENDPOINT_NAME = "CoarseGrainedScheduler"
}

而这里创建的DriverEndpoint是实现了ThreadSafeRpcEndpoint trait,而ThreadSafeRpcEndpoint 又继承了 RpcEndpoint trait。Spark Rpc通信请移步于此

// CoarseGrainedSchedulerBackend.scala line 303

protected def createDriverEndpoint(properties: Seq[(String, String)]): DriverEndpoint = {
  new DriverEndpoint(rpcEnv, properties)
}

了解了生命周期之后,我们再回到 DriverEndpoint的部分。当DriverEndpoint创建之后,按照生命周期顺序,下一步就会调用onStart方法。

在onStart方法中,是每隔一段时间发送ReviveOffers 类型的消息,默认是每隔1秒。

// CoarseGrainedSchedulerBackend.scala line 95

override def onStart() {
  // Periodically revive offers to allow delay scheduling to work
  val reviveIntervalMs = conf.getTimeAsMs("spark.scheduler.revive.interval", "1s")

  reviveThread.scheduleAtFixedRate(new Runnable {
    override def run(): Unit = Utils.tryLogNonFatalError {
      Option(self).foreach(_.send(ReviveOffers))
    }
  }, 0, reviveIntervalMs, TimeUnit.MILLISECONDS)
}
// org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessage.scala line 67
// Internal messages in driver
case object ReviveOffers extends CoarseGrainedClusterMessage

这个类型的消息挺有意思,上面注释也很明确的告知,driver内部的消息

可以这样理解,Driver启动之后,会定时掐一下自己,给自己发送一个ReviveOffers的消息。这里的send方法是self的方法,回忆Spark Rpc通信的内容,可知self是一个RpcEndPointRef对象。

 

那这个消息发送之后,由DriverEndpoint的receive接收后处理

// CoarseGrainedSchedulerBackend.scala line 122

case ReviveOffers =>
  makeOffers()

至此,DriverEndpoint创建完成,进入运行时状态了。同时CoarseGrainedSchedulerBackend也创建完成。具体的运行时,咱后续再分析。

 

下篇将介绍Driver向Master注册信息

 

共有 人打赏支持
粉丝 21
博文 103
码字总数 64503
×
柯里昂
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: