文档章节

Spark 之SparkContext 源码精读1

柯里昂
 柯里昂
发布于 2016/04/04 19:29
字数 1465
阅读 380
收藏 1

前文可知,SparkContext是Spark的核心中的核心。

SparkContext 是Spark功能的入口,表示与Spark 集群的连接。用于创建 RDD、广播变量、累加器变量;

 

接下来,咱们一起从源码开始,强烈建议读者把源码下载下来,和我一起分析源码。

引用老师的话,"源码能说明一切问题"

创建核心组件:TaskScheduler和SchedulerBackend

// SparkContext.scala line 521
// Create and start the scheduler
val (sched, ts) = SparkContext.createTaskScheduler(this, master)
// SparkContext.scala line 2592
  /**
   * Create a task scheduler based on a given master URL.
   * Return a 2-tuple of the scheduler backend and the task scheduler.
   */
  private def createTaskScheduler(
      sc: SparkContext,
      master: String): (SchedulerBackend, TaskScheduler) = {
    import SparkMasterRegex._

    // When running locally, don't try to re-execute tasks on failure.
    val MAX_LOCAL_TASK_FAILURES = 1

    master match {
      case "local" =>
        val scheduler = new TaskSchedulerImpl(sc, MAX_LOCAL_TASK_FAILURES, isLocal = true)
        val backend = new LocalBackend(sc.getConf, scheduler, 1)
        scheduler.initialize(backend)
        (backend, scheduler)
        
// ... 若干模式匹配
// line 2629
      case SPARK_REGEX(sparkUrl) =>
        val scheduler = new TaskSchedulerImpl(sc)
        val masterUrls = sparkUrl.split(",").map("spark://" + _)
        val backend = new SparkDeploySchedulerBackend(scheduler, sc, masterUrls)
        scheduler.initialize(backend)
        (backend, scheduler)
        
// ... 若干模式匹配

      case zkUrl if zkUrl.startsWith("zk://") =>
        logWarning("Master URL for a multi-master Mesos cluster managed by ZooKeeper should be " +
          "in the form mesos://zk://host:port. Current Master URL will stop working in Spark 2.0.")
        createTaskScheduler(sc, "mesos://" + zkUrl)

      case _ =>
        throw new SparkException("Could not parse Master URL: '" + master + "'")
    }
  }
}

这里着重看 SPARK_REGEX模式 。line 2630。首先创建 TaskSchedulerImpl 

 

// TaskSchedulerImpl.scala line 110
// default scheduler is FIFO
private val schedulingModeConf = conf.get("spark.scheduler.mode", "FIFO")
val schedulingMode: SchedulingMode = try {
  SchedulingMode.withName(schedulingModeConf.toUpperCase)
} catch {
  case e: java.util.NoSuchElementException =>
    throw new SparkException(s"Unrecognized spark.scheduler.mode: $schedulingModeConf")
}

在创建TaskScheduler时,指定了调度模型,默认是FIFO:先入先出。

其他的变量都是初始化,暂先不细究。

其他的方法都是需要从实例的对象中去调用的,也暂不细究。

 

此时TaskSchedulerImpl创建成功,然后将创建的TaskScheduler实例作为构造参数,创建SchedulerBackend。见SparkContext.scala line 2632。

此时创建的SchedulerBackend实例实际上是 SparkDeploySchedulerBackend 类。

// SparkDeploySchedulerBackend.scala line 30
private[spark] class SparkDeploySchedulerBackend(
    scheduler: TaskSchedulerImpl,
    sc: SparkContext,
    masters: Array[String])
  extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv)
  with AppClientListener
  with Logging

看SparkDeploySchedulerBackend 的定义可知,是继承自父类CoarseGrainedSchedulerBackend。让我们看看CoarseGrainedSchedulerBackend的定义。

// CoarseGrainedSchedulerBackend.scala line 31
/**
 * A scheduler backend that waits for coarse grained executors to connect to it through Akka.
 * This backend holds onto each executor for the duration of the Spark job rather than relinquishing
 * executors whenever a task is done and asking the scheduler to launch a new executor for
 * each new task. Executors may be launched in a variety of ways, such as Mesos tasks for the
 * coarse-grained Mesos mode or standalone processes for Spark's standalone deploy mode
 * (spark.deploy.*).
 */
private[spark]
class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: RpcEnv)
  extends ExecutorAllocationClient with SchedulerBackend with Logging

从上面的注释可知,这个类是粗粒度【coarse grained】的实现。扩展下,fine-grained【细粒度】的实现是MesosSchedulerBackend

大家应该知道,在创建一个对象实例是,会先执行父类的构造函数。这里,创建SparkDeploySchedulerBackend会先调用CoarseGrainedSchedulerBackend的构造。

至此SchedulerBackend创建成功。

值得注意的是:SchedulerBackend实例中有一个TaskScheduler类型的成员变量。后续有一些TaskScheduler关于调度的方法,会在SchedulerBackend中被调用。

紧接着,看上面代码,调用了

// SparkContext.scala line 2633
scheduler.initialize(backend)

上述代码清楚的说明了,将SchedulerBackend作为参数传进来。

至此,咱们可以简单的这么认为,SchedulerBackend中Backend 实际上就是指TaskScheduler。SchedulerBackend是TaskScheduler的后端。

再来看下TaskSchedulerImpl的initialize方法

// TaskSchedulerImpl.scala line 125
def initialize(backend: SchedulerBackend) {
  this.backend = backend
  // temporarily set rootPool name to empty
  rootPool = new Pool("", schedulingMode, 0, 0)
  schedulableBuilder = {
    schedulingMode match {
      case SchedulingMode.FIFO =>
        new FIFOSchedulableBuilder(rootPool)
      case SchedulingMode.FAIR =>
        new FairSchedulableBuilder(rootPool, conf)
    }
  }
  schedulableBuilder.buildPools()
}

很显然,TaskScheduler将SchedulerBackend实例作为成员变量保存了。后续源码分析中可见,此成员变量被频繁的使用。

同时,初始化了一个调度池,然后根据上面提到的调度模式初始化了调度建造器,之后就创建了调度池。调度模式后面会单独提到。尽请期待。

 

createTaskScheduler调用之后,会返回一个Tuple,具体是SchedulerBackend[SparkDeploySchedulerBackend] 和 TaskScheduler[TaskSchedulerImpl]

之后,将SparkContext作为参数,创建DAGScheduler。

DAG是Directed Acyclic Graph的缩写,是指有向无环图。

// SparkContext.scala line 521
// Create and start the scheduler ,实际上,方法调用完,只是创建了实例,并没有start,start是在 line 530
val (sched, ts) = SparkContext.createTaskScheduler(this, master)
_schedulerBackend = sched
_taskScheduler = ts
_dagScheduler = new DAGScheduler(this) // line 525 
_heartbeatReceiver.ask[Boolean](TaskSchedulerIsSet)

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

大家可能会问,明明上面创建了TaskScheduler和SchedulerBackend,却有没有马上start,是因为什么呢?

请看 line 530 上面的注释,已经很清楚的说明了原因。那么让我们看看DAGScheduler的构造中做了什么?

// DAGScheduler.scala line 131
def this(sc: SparkContext) = this(sc, sc.taskScheduler)
// DAGScheduler.scala line 121
def this(sc: SparkContext, taskScheduler: TaskScheduler) = {
  this(
    sc,
    taskScheduler,
    sc.listenerBus,
    sc.env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster],
    sc.env.blockManager.master,
    sc.env)
}

    最后的目的,就是为了将TaskScheduler传进去

// DAGScheduler.scala line 110
private[spark]
class DAGScheduler(
    private[scheduler] val sc: SparkContext,
    private[scheduler] val taskScheduler: TaskScheduler,
    listenerBus: LiveListenerBus,
    mapOutputTracker: MapOutputTrackerMaster,
    blockManagerMaster: BlockManagerMaster,
    env: SparkEnv,
    clock: Clock = new SystemClock())
  extends Logging

并在后续构造中,将DAGScheduler赋值给TaskScheduler的成员变量

// DAGScheduler.scala line 185
taskScheduler.setDAGScheduler(this)

由此可见,TaskScheduler和DAGScheduler相互之间是有引用的。

那这两个Scheduler之间有什么区别呢?

DAGScheduler可以认为是总监,他是负责阶段性进展的管控。

TaskScheduler可以认为是监工,负责具体的一个任务的进度。

比如一个项目有一期二期。一期下面有A、B、C,二期下面有D、E、F。

DAGScheduler是管控项目每一期的进展的。也就是一期、二期项目顺利完成。

而TaskScheduler是管控每一期中的子任务的进展的。也就是一期中的A、B、C。二期中的D、E、F 任务顺利完成。

当然顺利完成也代表着,如果遇到任务中断了,需要调整,重新开始。即失败重试机制。

而这两者的共同点是,他们都是负责不同粒度的任务顺利完成的。换言之,他们是不管完成任务所需资源从哪来的。

 

当DAGScheduler也创建完成后,SparkContext的3大核心对象已创建完成;分别是TaskScheduler,SchedulerBackend,DAGScheduler。

 

2016-04-04 ,时间不早了,今天就更新到此。

 

下篇介绍Driver的创建

 

© 著作权归作者所有

共有 人打赏支持
柯里昂
粉丝 25
博文 173
码字总数 82699
作品 0
徐汇
技术主管
私信 提问
加载中

评论(1)

柯里昂
柯里昂
DAGScheduler中只包含TaskScheduler的引用,没有SchedulerBackend的引用,再次说明了,SchedulerBackend中Scheduler是指TaskScheduler
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

没有更多内容

加载失败,请刷新页面

加载更多

 SQL与NoSQL数据库入门基础知识详解

这几年的大数据热潮带动了一激活了一大批hadoop学习爱好者。有自学hadoop的,有报名培训班学习的。所有接触过hadoop的人都知道,单独搭建hadoop里每个组建都需要运行环境、修改配置文件测试等...

左手的倒影
9分钟前
0
0
Jenkinsfile

node { wrap([$class: 'BuildUser']) { def user = env.BUILD_USER_ID println user }}def InPutCheckBranch(){ timeout(time:10, unit:'SECONDS') { I......

whhbb
14分钟前
0
0
如何去设计前端框架能力?星巴克消息开放项目从0到1,从点到面的思考

本文由淘宝前端工程师罗嗣分享,主要讲述了作者在星巴克消息开放项目中的总结和思考,希望对大家有帮助,让业务分享更加有价值。 摘要 从满足星巴克项目需求单点出发,发散到从点到面的思考。...

阿里云官方博客
19分钟前
0
0
如何在多Kubernetes集群和多租户环境中使用Prometheus监控

过去十年里,我一直在虚拟化和云计算领域工作,在VMware、Mercury Interactive和Oracle领导产品管理工作,并格外关注DevOps方向。直到2014年,我开始认真关注并研究容器。当2015年Kubernete...

RancherLabs
21分钟前
0
0
如何在Chrome浏览器中启动deviceready事件(尝试调试phonegap项目)?

我正在开发PhoneGap应用程序,我希望能够在Chrome中调试它,而不是在电话上调试。但是,我在onGetReady()函数中初始化我的代码,该函数在PhoneGap触发“deviceready”事件时触发。由于Chr...

kisshua
今天
10
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部