如何進(jìn)行DAGScheduler源碼解讀

如何進(jìn)行DAGScheduler源碼解讀，針對(duì)這個(gè)問(wèn)題，這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答，希望可以幫助更多想解決這個(gè)問(wèn)題的小伙伴找到更簡(jiǎn)單易行的方法。

當(dāng)構(gòu)建完TaskScheduler之后，我們需要構(gòu)建DAGScheduler這個(gè)核心對(duì)象：

進(jìn)入其構(gòu)造函數(shù)中：

可以看出構(gòu)建DAGScheduler實(shí)例的時(shí)候需要把TaskScheduler實(shí)例對(duì)象作為參數(shù)傳入。

LiveListenerBus：

BlockManagerMaster：

通過(guò)閱讀代碼，我們可以發(fā)現(xiàn)DAGScheduler實(shí)例化的時(shí)候，調(diào)用了initializeEventProcessActor()方法

private def initializeEventProcessActor() {  // blocking the thread until supervisor is started, which ensures eventProcessActor is  // not null before any job is submitted  // 阻塞當(dāng)前線(xiàn)程，等待supervisor啟動(dòng)，這樣可以確保Job提交時(shí)，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    // 設(shè)置taskScheduler對(duì)DAGScheduler的引用句柄。在此處設(shè)置保證了Job提交時(shí)候    // eventProcessActor已經(jīng)準(zhǔn)備就緒    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()
  }
}

可以看出核心在于實(shí)例化eventProcessActor對(duì)象，eventProcessActor會(huì)負(fù)責(zé)接收和發(fā)送DAGScheduler的消息，是DAGScheduler的通信載體。

關(guān)于如何進(jìn)行DAGScheduler源碼解讀問(wèn)題的解答就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，如果你還有很多疑惑沒(méi)有解開(kāi)，可以關(guān)注億速云行業(yè)資訊頻道了解更多相關(guān)知識(shí)。