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

創(chuàng)新互聯(lián)建站專注于鎮(zhèn)原網(wǎng)站建設(shè)服務(wù)及定制，我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗。 熱誠為您提供鎮(zhèn)原營銷型網(wǎng)站建設(shè)，鎮(zhèn)原網(wǎng)站制作、鎮(zhèn)原網(wǎng)頁設(shè)計、鎮(zhèn)原網(wǎng)站官網(wǎng)定制、小程序制作服務(wù)，打造鎮(zhèn)原網(wǎng)絡(luò)公司原創(chuàng)品牌,更為您提供鎮(zhèn)原網(wǎng)站排名全網(wǎng)營銷落地服務(wù)。

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

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

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

LiveListenerBus：

BlockManagerMaster：

通過閱讀代碼，我們可以發(fā)現(xiàn)DAGScheduler實例化的時候，調(diào)用了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    // 設(shè)置taskScheduler對DAGScheduler的引用句柄。在此處設(shè)置保證了Job提交時候    // eventProcessActor已經(jīng)準備就緒    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會負責接收和發(fā)送DAGScheduler的消息，是DAGScheduler的通信載體。

關(guān)于如何進行DAGScheduler源碼解讀問題的解答就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道了解更多相關(guān)知識。