文档章节

Spark 之SparkContext 源码精读1

柯里昂
 柯里昂
发布于 2016/04/04 19:29
字数 1465
阅读 376
收藏 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的创建

 

© 著作权归作者所有

共有 人打赏支持
柯里昂
粉丝 24
博文 168
码字总数 81139
作品 0
徐汇
技术主管
加载中

评论(1)

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

扣丁学堂
昨天
0
0
【Spark亚太研究院系列丛书】Spark实战高手之路-第一章 构建Spark集群(第四步)(1)

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

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

没有更多内容

加载失败,请刷新页面

加载更多

下一页

vue+element-ui操作删除(单行和批量删除)

页面展示: <template><!-- 表格内容 --><el-table :data="packData" border style="width: 100%" ref="multipleTable" @selection-change="handleSelectionChange"><el-tab......

琴妹
4分钟前
0
0
基于vue(element ui) + ssm + shiro 的权限框架

zhcc 基于vue(element ui) + ssm + shiro 的权限框架 引言 心声 现在的Java世界,各种资源很丰富,不得不说,从分布式,服务化,orm,再到前端控制,权限等等玲琅满目,网上有句话说,语言框架...

DarrenHu_吴邪
12分钟前
0
1
数据库水平切分(MyCat分片)

范围分片 io.mycat.route.function.AutoPartitionByLong 自动范围分片 Function名称:rang-long(配置文件默认) 枚举分片 io.mycat.route.function.PartitionByFileMap 枚举分片 Funtion名称...

这很耳东先生
13分钟前
0
0
读《HeadFirst设计模式》笔记之外观模式

外观模式:提供了一个统一的接口,用来访问子系统中的一群接口。外观定义了一个高层接口,让子系统更容易使用。 举个栗子: 建了一个家庭影院,但是每次享受家庭影院时,你发现需要执行 将灯...

suyain
14分钟前
0
0
MongoDB分片配置

简单注解: mongos 路由进程, 应用程序接入mongos再查询到具体分片,监听端口默认27017 config server 路由表服务, 每一台都具有全部chunk的路由信息 shard为数据存储分片, 每一片都可以是...

LUIS1983
21分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部