文档章节

spark内核揭秘-07-DAGScheduler源码解读初体验

stark_summer
 stark_summer
发布于 2015/01/19 20:57
字数 414
阅读 69
收藏 2

当构建完TaskScheduler之后,我们需要构建DAGScheduler这个核心对象:


进入其构造函数中:





可以看出构建DAGScheduler实例的时候需要把TaskScheduler实例对象作为参数传入。

LiveListenerBus:


MapOutputTrackerMaster:


BlockManagerMaster:


通过阅读代码,我们可以发现DAGScheduler实例化的时候,调用了initializeEventProcessActor()方法

private def initializeEventProcessActor() {  // blocking the thread until supervisor is started, which ensures eventProcessActor is  // not null before any job is submitted  // 阻塞当前线程,等待supervisor启动,这样可以确保Job提交时,eventProcessActor not null  implicit val timeout = Timeout(30 seconds)  val initEventActorReply =    dagSchedulerActorSupervisor ? Props(new DAGSchedulerEventProcessActor(this))  eventProcessActor = Await.result(initEventActorReply, timeout.duration).
    asInstanceOf[ActorRef]
}

initializeEventProcessActor()

DAGSchedulerEventProcessActor:

private[scheduler] class DAGSchedulerEventProcessActor(dagScheduler: DAGScheduler)  extends Actor with Logging {  override def preStart() {    // set DAGScheduler for taskScheduler to ensure eventProcessActor is always    // valid when the messages arrive    // 设置taskScheduler对DAGScheduler的引用句柄。在此处设置保证了Job提交时候    // eventProcessActor已经准备就绪    dagScheduler.taskScheduler.setDAGScheduler(dagScheduler)
  }  /**   * The main event loop of the DAG scheduler.   */  def receive = {    case JobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite, listener, properties) =>
      dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite,        listener, properties)    case StageCancelled(stageId) =>
      dagScheduler.handleStageCancellation(stageId)    case JobCancelled(jobId) =>
      dagScheduler.handleJobCancellation(jobId)    case JobGroupCancelled(groupId) =>
      dagScheduler.handleJobGroupCancelled(groupId)    case AllJobsCancelled =>
      dagScheduler.doCancelAllJobs()    case ExecutorAdded(execId, host) =>
      dagScheduler.handleExecutorAdded(execId, host)    case ExecutorLost(execId) =>
      dagScheduler.handleExecutorLost(execId, fetchFailed = false)    case BeginEvent(task, taskInfo) =>
      dagScheduler.handleBeginEvent(task, taskInfo)    case GettingResultEvent(taskInfo) =>
      dagScheduler.handleGetTaskResult(taskInfo)    case completion @ CompletionEvent(task, reason, _, _, taskInfo, taskMetrics) =>
      dagScheduler.handleTaskCompletion(completion)    case TaskSetFailed(taskSet, reason) =>
      dagScheduler.handleTaskSetFailed(taskSet, reason)    case ResubmitFailedStages =>
      dagScheduler.resubmitFailedStages()
  }  override def postStop() {    // Cancel any active jobs in postStop hook    dagScheduler.cleanUpAfterSchedulerStop()
  }
}


可以看出核心在于实例化eventProcessActor对象,eventProcessActor会负责接收和发送DAGScheduler的消息,是DAGScheduler的通信载体。

© 著作权归作者所有

共有 人打赏支持
stark_summer
粉丝 59
博文 75
码字总数 51050
作品 0
朝阳
CEO
Spark源码分析:多种部署方式之间的区别与联系(1)

 从官方的文档我们可以知道,Spark的部署方式有很多种:local、Standalone、Mesos、YARN.....不同部署方式的后台处理进程是不一样的,但是如果我们从代码的角度来看,其实流程都差不多。  ...

Ryan-瑞恩
2015/09/02
0
0
Spark2.1.0之基础知识

在阅读本文之前,读者最好已经阅读了《Spark2.1.0之初识Spark》一文,本文将对Spark的基础知识进行介绍。但在此之前,读者先跟随本人来一次简单的时光穿梭,最后还将对Java与Scala在语言上进...

beliefer
05/24
0
0
教你如何成为Spark大数据高手

Spark目前被越来越多的企业使用,和Hadoop一样,Spark也是以作业的形式向集群提交任务,那么如何成为Spark大数据高手?下面就来个深度教程。 分享之前我还是要推荐下我自己创建的大数据学习交...

风火数据
05/20
0
0
【Spark亚太研究院系列丛书】Spark实战高手之路-第3章Spark架构设计与编程模型第1节②

三、你为什么需要Spark; 你需要Spark的十大理由: 1,Spark是可以革命Hadoop的目前唯一替代者,能够做Hadoop做的一切事情,同时速度比Hadoop快了100倍以上: Logistic regression in Hadoo...

Spark亚太研究院
2014/12/16
0
1
Spark 学习资源收集【Updating】

(一)spark 相关安装部署、开发环境 1、Spark 伪分布式 & 全分布式 安装指南 http://my.oschina.net/leejun2005/blog/394928 2、Apache Spark探秘:三种分布式部署方式比较 http://dongxic...

大数据之路
2014/09/08
0
1

没有更多内容

加载失败,请刷新页面

加载更多

MySQL autocommit探究

-- sessionA:tx_isolation=REPEATABLE-READmysql> select connection_id();+-----------------+| connection_id() |+-----------------+| 28 |+-----------------+......

安小乐
3分钟前
0
0
c++多线程锁 Mutex  自动判断死锁

c++多线程锁可以使用absl::Mutex std::mutex这两种,下面是demo代码。 使用absl:Mutex的时候打印: [mutex.cc : 1338] RAW: Cycle: [mutex.cc : 1352] RAW: mutex@0x683b68 stack: @ 0x43856......

青黑
22分钟前
0
0
Blockathon2018(成都站)比赛落幕,留给我们这些区块链应用思考

9月14日,HiBlock区块链社区主办的第二届Blockathon在成都菁融国际广场成功举行,30名参赛者分为5支队伍在48小时内完成区块链项目的创意、开发及路演,经过紧张的开发及现场评选,最终币托(...

HiBlock
27分钟前
0
0
71.告警系统主脚本 配置文件 监控项目

20.20 告警系统主脚本(main.sh) 20.21 告警系统配置文件 20.22 告警系统监控项目 20.20 告警系统主脚本(main.sh): ~1.约定:把以后所有的shell脚本放在/usr/local/sbin下,也方便我们查...

王鑫linux
34分钟前
0
0
装饰者模式

装饰者模式 Q:何为装饰模式? ()地给一个对象添加一些额外的(),并且()时,并不影响原对象。扩展功能来说,装饰器模式相比生成子类更为灵活。 Q:使用场景? 1.想要在不影响其他对象的情况下...

阿元
54分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部