DAGScheduler is the scheduling layer of Apache Spark that implements stage-oriented scheduling, i.e. after an RDD action has been called it becomes a job that is then transformed into a set of stages that are submitted as TaskSets for execution (see Execution Model).

The fundamental concepts of DAGScheduler are jobs and stages (refer to Jobs and Stages respectively) that it tracks through internal registries and counters.

DAGScheduler works solely on the driver and is created as part of SparkContext’s initialization (right after TaskScheduler and SchedulerBackend are ready).

DAGScheduler does three things in Spark (thorough explanations follow):

It computes a directed acyclic graph (DAG) of stages for each job, keeps track of which RDDs and stage outputs are materialized, and finds a minimal schedule to run jobs. It then submits stages to TaskScheduler.

In addition to coming up with the execution DAG, DAGScheduler also determines the preferred locations to run each task on, based on the current cache status, and passes the information to TaskScheduler.

Furthermore, it handles failures due to shuffle output files being lost, in which case old stages may need to be resubmitted. Failures within a stage that are not caused by shuffle file loss are handled by the TaskScheduler itself, which will retry each task a small number of times before cancelling the whole stage.

DAGScheduler uses an event queue architecture in which a thread can post DAGSchedulerEvent events, e.g. a new job or stage being submitted, that DAGScheduler reads and executes sequentially. See the section Internal Event Loop - dag-scheduler-event-loop.

DAGScheduler runs stages in topological order.

DAGScheduler needs SparkContext, Task Scheduler, LiveListenerBus, MapOutputTracker and Block Manager to work. However, at the very minimum, DAGScheduler needs SparkContext only (and asks SparkContext for the other services).

DAGScheduler reports metrics about its execution (refer to the section Metrics).

When DAGScheduler schedules a job as a result of executing an action on a RDD or calling SparkContext.runJob() method directly, it spawns parallel tasks to compute (partial) results per partition.

getOrCreateParentStages computes ShuffleDependency immediate parents and gets or creates a shuffle map stage for each ShuffleDependency.

getShuffleDependencies…​TK

DAGScheduler requires a SparkContext, TaskScheduler, LiveListenerBus, MapOutputTrackerMaster, BlockManagerMaster, SparkEnv, and a Clock.

When created, DAGScheduler does the following (in order):

At the very end of the initialization, DAGScheduler starts eventProcessLoop.

listenerBus is a LiveListenerBus to post scheduling events and is passed in when DAGScheduler is created.

DAGScheduler uses internal registries and counters for managing active jobs and stages.

executorHeartbeatReceived posts a SparkListenerExecutorMetricsUpdate (to listenerBus) and informs BlockManagerMaster that blockManagerId block manager is alive (by posting BlockManagerHeartbeat).

cleanupStateForJobAndIndependentStages cleans up the state for job and any stages that are not part of any other job.

cleanupStateForJobAndIndependentStages looks the job up in the internal jobIdToStageIds registry.

If no stages are found, the following ERROR is printed out to the logs:

Oterwise, cleanupStateForJobAndIndependentStages uses stageIdToStage registry to find the stages (the real objects not ids!).

For each stage, cleanupStateForJobAndIndependentStages reads the jobs the stage belongs to.

If the job does not belong to the jobs of the stage, the following ERROR is printed out to the logs:

If the job was the only job for the stage, the stage (and the stage id) gets cleaned up from the registries, i.e. runningStages, shuffleIdToMapStage, waitingStages, failedStages and stageIdToStage.

While removing from runningStages, you should see the following DEBUG message in the logs:

While removing from waitingStages, you should see the following DEBUG message in the logs:

While removing from failedStages, you should see the following DEBUG message in the logs:

After all cleaning (using stageIdToStage as the source registry), if the stage belonged to the one and only job, you should see the following DEBUG message in the logs:

The job is removed from jobIdToStageIds, jobIdToActiveJob, activeJobs registries.

The final stage of the job is removed, i.e. ResultStage or ShuffleMapStage.

markMapStageJobAsFinished marks map stage jobs finished and notifies Spark listeners.

Internally, markMapStageJobAsFinished marks the zeroth partition finished and increases the number of tasks finished in job.

The job listener is notified about the 0th task succeeded.

The state of the job and independent stages are cleaned up.

Ultimately, SparkListenerJobEnd is posted to LiveListenerBus (as listenerBus) for the job, the current time (in millis) and JobSucceeded job result.

clearCacheLocs clears the internal cache of the partition locations of all RDDs.

The internal failJobAndIndependentStages method fails the input job and all the stages that are only used by the job.

Internally, failJobAndIndependentStages uses jobIdToStageIds internal registry to look up the stages registered for the job.

If no stages could be found, you should see the following ERROR message in the logs:

Otherwise, for every stage, failJobAndIndependentStages finds the job ids the stage belongs to.

If no stages could be found or the job is not referenced by the stages, you should see the following ERROR message in the logs:

Only when there is exactly one job registered for the stage and the stage is in RUNNING state (in runningStages internal registry), TaskScheduler is requested to cancel the stage’s tasks and marks the stage finished.

submitJob creates a JobWaiter and posts a JobSubmitted event.

Internally, submitJob does the following:

You may see a IllegalArgumentException thrown when the input partitions references partitions not in the input rdd:

cancelJobGroup merely posts a StageCancelled event to the DAGScheduler’s Internal Event Bus.

cancelJobGroup prints the following INFO message to the logs followed by posting a JobGroupCancelled event to the DAGScheduler’s Internal Event Bus.

cancelAllJobs merely posts a AllJobsCancelled event to the DAGScheduler’s Internal Event Bus.

taskStarted merely posts a BeginEvent event to the DAGScheduler’s Internal Event Bus.

taskGettingResult merely posts a GettingResultEvent event to the DAGScheduler’s Internal Event Bus.

taskEnded merely posts a CompletionEvent event to the DAGScheduler’s Internal Event Bus.

submitMapStage posts a MapStageSubmitted event to the DAGScheduler’s Internal Event Bus and returns the JobWaiter with one task only and a result handler that will call the callback function.

submitMapStage increments nextJobId for the job id.

taskSetFailed simply posts a TaskSetFailed to DAGScheduler’s Internal Event Bus.

executorLost simply posts a ExecutorLost event to DAGScheduler’s Internal Event Bus.

executorAdded simply posts a ExecutorAdded event to DAGScheduler’s Internal Event Bus.

cancelJob prints the following INFO message and posts a JobCancelled to DAGScheduler’s Internal Event Bus.

runJob submits an action job to the DAGScheduler and waits for a result.

Internally, runJob executes submitJob and then waits until a result comes using JobWaiter.

When the job succeeds, you should see the following INFO message in the logs:

When the job fails, you should see the following INFO message in the logs and the exception (that led to the failure) is thrown.

abortStage is an internal method that finds all the active jobs that depend on the failedStage stage and fails them.

Internally, abortStage looks the failedStage stage up in the internal stageIdToStage registry and exits if there the stage was not registered earlier.

If it was, abortStage finds all the active jobs (in the internal activeJobs registry) with the final stage depending on the failedStage stage.

At this time, the completionTime property (of the failed stage’s StageInfo) is assigned to the current time (millis).

All the active jobs that depend on the failed stage (as calculated above) and the stages that do not belong to other jobs (aka independent stages) are failed (with the failure reason being "Job aborted due to stage failure: [reason]" and the input exception).

If there are no jobs depending on the failed stage, you should see the following INFO message in the logs:

stageDependsOn compares two stages and returns whether the stage depends on target stage (i.e. true) or not (i.e. false).

Internally, stageDependsOn walks through the graph of RDDs of the input stage. For every RDD in the RDD’s dependencies (using RDD.dependencies) stageDependsOn adds the RDD of a NarrowDependency to a stack of RDDs to visit while for a ShuffleDependency it getOrCreateShuffleMapStage for the dependency and the stage's first job id that it later adds to a stack of RDDs to visit if the map stage is ready, i.e. all the partitions have shuffle outputs.

After all the RDDs of the input stage are visited, stageDependsOn checks if the target's RDD is among the RDDs of the stage, i.e. whether the stage depends on target stage.

eventProcessLoop is DAGScheduler’s event bus to which Spark (by submitJob) posts jobs to schedule their execution. Later on, TaskSetManager talks back to DAGScheduler to inform about the status of the tasks using the same "communication channel".

It allows Spark to release the current thread when posting happens and let the event loop handle events on a separate thread - asynchronously.

…​IMAGE…​FIXME

submitWaitingStages method checks for waiting or failed stages that could now be eligible for submission.

When executed, you should see the following TRACE messages in the logs:

The method clears the internal waitingStages set with stages that wait for their parent stages to finish.

It goes over the waiting stages sorted by job ids in increasing order and calls submitStage method.

submitStage is an internal method that DAGScheduler uses to submit the input stage or its missing parents (if there any).

submitStage recursively submits any missing parents of the stage.

Internally, submitStage first finds the earliest-created ActiveJob that needs the stage.

You should see the following DEBUG message in the logs:

Only when the stage is not in waiting (waitingStages), running (runningStages) or failed states submitStage proceeds.

The list of missing parent stages of the stage is calculated (sorted by their ids) and the following DEBUG message shows up in the logs:

When the stage has no parent stages missing, you should see the following INFO message in the logs:

The stage is submitted. That finishes the stage submission.

If however there are missing parent stages for the stage, all parent stages are submitted (by id in increasing order), and the stage is added to waitingStages stages.

In case when submitStage could find no active job for the stage, it aborts the stage with the reason:

getMissingParentStages calculates missing parent map stages for the input stage.

It starts with the stage’s target RDD (as stage.rdd). If there are uncached partitions, it traverses the dependencies of the RDD (as RDD.dependencies) that can be the instances of ShuffleDependency or NarrowDependency.

For each ShuffleDependency, the method searches for the corresponding ShuffleMapStage (using getShuffleMapStage) and if unavailable, the method adds it to a set of missing (map) stages.

It continues traversing the chain for each NarrowDependency (using Dependency.rdd).

A single stage can be re-executed in multiple attempts due to fault recovery. The number of attempts is configured (FIXME).

If TaskScheduler reports that a task failed because a map output file from a previous stage was lost, the DAGScheduler resubmits that lost stage. This is detected through a CompletionEvent with FetchFailed, or an ExecutorLost event. DAGScheduler will wait a small amount of time to see whether other nodes or tasks fail, then resubmit TaskSets for any lost stage(s) that compute the missing tasks.

Please note that tasks from the old attempts of a stage could still be running.

A stage object tracks multiple StageInfo objects to pass to Spark listeners or the web UI.

The latest StageInfo for the most recent attempt for a stage is accessible through latestInfo.

DAGScheduler tracks which RDDs are cached to avoid recomputing them and likewise remembers which shuffle map stages have already produced output files to avoid redoing the map side of a shuffle.

DAGScheduler is only interested in cache location coordinates, i.e. host and executor id, per partition of an RDD.

If the storage level of an RDD is NONE, there is no caching and hence no partition cache locations are available. In such cases, whenever asked, DAGScheduler returns a collection with empty-location elements for each partition. The empty-location elements are to mark uncached partitions.

Otherwise, a collection of RDDBlockId instances for each partition is created and spark-BlockManagerMaster.adoc[BlockManagerMaster] is asked for locations (using BlockManagerMaster.getLocations). The result is then mapped to a collection of TaskLocation for host and executor id.

DAGScheduler computes where to run each task in a stage based on the preferred locations of its underlying RDDs, or the location of cached or shuffle data.

See SPARK-9850 Adaptive execution in Spark for the design document. The work is currently in progress.

DAGScheduler.submitMapStage method is used for adaptive query planning, to run map stages and look at statistics about their outputs before submitting downstream stages.

DAGScheduler uses the following ScheduledThreadPoolExecutors (with the policy of removing cancelled tasks from a work queue at time of cancellation):

They are created using ThreadUtils.newDaemonSingleThreadScheduledExecutor method that uses Guava DSL to instantiate a ThreadFactory.

submitMissingTasks is a private method that…​FIXME

When executed, it prints the following DEBUG message out to the logs:

The stage’s pendingPartitions internal field is cleared (it is later filled out with the partitions to run tasks for).

The stage is asked for partitions to compute (see findMissingPartitions in Stages).

The method adds the stage to runningStages internal registry.

The stage is told to be started to OutputCommitCoordinator (using outputCommitCoordinator.stageStart)

The mapping between task ids and task preferred locations is computed (see getPreferredLocs - Computing Preferred Locations for Tasks and Partitions).

A new stage attempt is created (using Stage.makeNewStageAttempt).

SparkListenerStageSubmitted is posted.

The stage is serialized and broadcast to workers using SparkContext.broadcast method, i.e. it is Serializer.serialize to calculate taskBinaryBytes - an array of bytes of (rdd, func) for ResultStage and (rdd, shuffleDep) for ShuffleMapStage.

When serializing the stage fails, the stage is removed from runningStages internal registry, the stage is aborted and the method stops.

At this point in time, the stage is on workers.

For each partition to compute for the stage, a collection of ShuffleMapTask for ShuffleMapStage or ResultTask for ResultStage is created.

Any issue with creating a task leads to aborting the stage and removing the stage from runningStages internal registry.

If there are tasks to launch (there are missing partitions in the stage), the following INFO and DEBUG messages are in the logs:

All tasks in the collection become a TaskSet for TaskScheduler.submitTasks.

In case of no tasks to be submitted for a stage, a DEBUG message shows up in the logs.

For ShuffleMapStage:

For ResultStage:

stop stops the internal dag-scheduler-message thread pool, dag-scheduler-event-loop, and TaskScheduler.

Spark’s DAGScheduler uses Spark Metrics System (via DAGSchedulerSource) to report metrics about internal status.

The name of the source is DAGScheduler.

It emits the following numbers:

The private updateAccumulators method merges the partial values of accumulators from a completed task into their "source" accumulators on the driver.

For each AccumulableInfo in the CompletionEvent, a partial value from a task is obtained (from AccumulableInfo.update) and added to the driver’s accumulator (using Accumulable.++= method).

For named accumulators with the update value being a non-zero value, i.e. not Accumulable.zero: