文档章节

Spark 之SparkContext 源码精读2

柯里昂
 柯里昂
发布于 2016/04/05 08:48
字数 858
阅读 186
收藏 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注册信息

 

© 著作权归作者所有

共有 人打赏支持
柯里昂
粉丝 24
博文 169
码字总数 81236
作品 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
【Spark亚太研究院系列丛书】Spark实战高手之路-第一章 构建Spark集群(第四步)(1)

第一步:通过Spark的shell测试Spark的工作 Step1:启动Spark集群,这一点在第三讲讲的极为细致,启动后的WebUI如下: Step2:启动Spark Shell: 此时可以通过如下Web控制台查看shell的情况: ...

Spark亚太研究院
2014/09/11
0
0

没有更多内容

加载失败,请刷新页面

加载更多

day92-20180918-英语流利阅读-待学习

健身最大的敌人不是懒惰,而是逞强 Daniel 2018-09-19 1.今日导读 还记得 2008 年北京奥运会运动员刘翔的退赛风波吗?那天几乎所有中国人都将视线聚焦在了鸟巢体育馆 110 米栏的项目上,迫不...

飞鱼说编程
25分钟前
2
0
70.shell的函数 数组 告警系统需求分析

20.16/20.17 shell中的函数 20.18 shell中的数组 20.19 告警系统需求分析 20.16/20.17 shell中的函数: ~1. 函数就是把一段代码整理到了一个小单元中,并给这个小单元起一个名字,当用到这段...

王鑫linux
今天
3
0
分布式框架spring-session实现session一致性使用问题

前言:项目中使用到spring-session来缓存用户信息,保证服务之间session一致性,但是获取session信息为什么不能再服务层获取? 一、spring-session实现session一致性方式 用户每一次请求都会...

WALK_MAN
今天
6
0
C++ yield()与sleep_for()

C++11 标准库提供了yield()和sleep_for()两个方法。 (1)std::this_thread::yield(): 线程调用该方法时,主动让出CPU,并且不参与CPU的本次调度,从而让其他线程有机会运行。在后续的调度周...

yepanl
今天
4
0
Java并发编程实战(chapter_3)(线程池ThreadPoolExecutor源码分析)

这个系列一直没再写,很多原因,中间经历了换工作,熟悉项目,熟悉新团队等等一系列的事情。并发课题对于Java来说是一个又重要又难的一大块,除非气定神闲、精力满满,否则我本身是不敢随便写...

心中的理想乡
今天
56
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部