Shulou(Shulou.com)05/31 Report--

How to interpret the DAGScheduler source code, in view of this problem, this article introduces the corresponding analysis and answers in detail, hoping to help more partners who want to solve this problem to find a more simple and easy way.

When we have finished building TaskScheduler, we need to build the core object DAGScheduler:

Enter its constructor:

You can see that when you build a DAGScheduler instance, you need to pass in the TaskScheduler instance object as a parameter.

LiveListenerBus:

BlockManagerMaster:

By reading the code, we can see that the initializeEventProcessActor () method is called when DAGScheduler is instantiated

Private def initializeEventProcessActor () {/ / blocking the thread until supervisor is started, which ensures eventProcessActor is / / not null before any job is submitted / / blocks the current thread and waits for supervisor to start, which ensures that when Job is submitted, 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 / / sets taskScheduler's reference handle to DAGScheduler. Setting here ensures that / / eventProcessActor is ready when Job is submitted. 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 ()}}

You can see that the core is to instantiate eventProcessActor objects, and eventProcessActor is responsible for receiving and sending messages from DAGScheduler, which is the communication carrier of DAGScheduler.

The answer to the question on how to interpret the DAGScheduler source code is shared here. I hope the above content can be of some help to everyone. If you still have a lot of doubts to be solved, you can follow the industry information channel to learn more about it.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.