Spark 之SparkContext 源码精读1
Spark 之SparkContext 源码精读1
柯里昂 发表于2年前
Spark 之SparkContext 源码精读1
  • 发表于 2年前
  • 阅读 348
  • 收藏 1
  • 点赞 1
  • 评论 1
摘要: SparkContext 3大核心对象的创建

前文可知,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的创建

 

标签: Spark SparkContext
共有 人打赏支持
粉丝 21
博文 103
码字总数 64643
评论 (1)
柯里昂
DAGScheduler中只包含TaskScheduler的引用,没有SchedulerBackend的引用,再次说明了,SchedulerBackend中Scheduler是指TaskScheduler
×
柯里昂
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: