文档章节

Spark 之SparkContext 源码精读2

柯里昂
 柯里昂
发布于 2016/04/05 08:48
字数 858
阅读 195
收藏 1

前文说到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注册信息

 

© 著作权归作者所有

共有 人打赏支持
柯里昂
粉丝 25
博文 173
码字总数 82699
作品 0
徐汇
技术主管
私信 提问
SPARK 源码分析技术分享(带bilibili视频)

SPARK 源码分析技术分享 (带bilibili视频) 【本站点正在持续更新中…2018-12-05…】 SPARK 1.6.0-cdh5.15.0 Hadoop 2.6.0-cdh5.15.0 spark-scala-maven 微信(技术交流) : thinktothings SPA...

thinktothings
12/02
0
0
1、Spark预编译版本下载安装与启动

1、下载 2、安装 3、hadoop 4、spark-shell scala> file.first() 5、Resilient Distributed Dataset 6、通过编译方式安装spark 轻量级高速集群计算。针对大规模的数据处理快速通用的引擎。比...

chenkangyao
2017/10/26
0
0
SparkContext 初始化内部原理

如果编写Spark程序,那么第⼀⾏代码就是new SparkContext().setMaster(“”).setAppName(“xx”),可以说SparkContext是整个Spark 计算的启动器,只有将sparkContext 启动起来,后续的关于调 ...

大数据之路
2012/11/12
0
0
Spark的运行架构分析(一)之架构概述

本博客转载自:https://blog.csdn.net/gamer_gyt/article/details/51822765 1:Spark的运行模式 2:Spark中的一些名词解释 3:Spark的运行基本流程 4:RDD的运行基本流程 一:Spark的运行模式...

lubin2016
04/18
0
0
扣丁学堂大数据培训Spark架构运行及优势详解

  今天扣丁学堂大数据培训给大家介绍一下关于大数据开发中Spark架构运行详解及其优势详解,首先spark是一种分布式的计算框架。类似于大数据开发中Hadoop生态圈的MapReduce,计算思想和MR非...

扣丁学堂
08/14
0
0

没有更多内容

加载失败,请刷新页面

加载更多

CentOS配置Tomcat监听80端口,虚拟主机

Tomcat更改默认端口为80 更改的配置文件是: /usr/local/tomcat/conf/server.xml [root@test-a ~]# vim /usr/local/tomcat/conf/server.xml # 找到 Connector port="8080" protocol="HTTP/1......

野雪球
25分钟前
2
0
《稻盛和夫经营学》读后感心得体会3180字范文

《稻盛和夫经营学》读后感心得体会3180字范文: 一代日本经营之圣稻盛和夫凭借刻苦勤奋的精神以及深植于佛教的商业道德准则,成为了“佛系”企业家的代表人物。在《稻盛和夫经营学》“领导人...

原创小博客
今天
1
0
java框架学习日志-5(常见的依赖注入)

依赖注入(dependency injection) 之前提到控制反转(Inversion of Control)也叫依赖注入,它们其实是一个东西,只是看的角度不同,这章详细说一下依赖注入。 依赖——指bean对象创建依赖于...

白话
今天
2
0
红外接收器驱动开发

背景:使用系统的红外遥控软件没有反应,然后以为自己接线错误,反复测试,结果烧坏了一个红外接收器,信号主板没有问题。所以自己开发了一个红外接收器的python驱动。接线参见https://my.os...

mbzhong
今天
2
0
ActiveMQ消息传送机制以及ACK机制详解

AcitveMQ是作为一种消息存储和分发组件,涉及到client与broker端数据交互的方方面面,它不仅要担保消息的存储安全性,还要提供额外的手段来确保消息的分发是可靠的。 一. ActiveMQ消息传送机...

watermelon11
今天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部