Scheduler

The scheduler turns user actions into stages, stages into tasks, and tasks into resource requests on a cluster. It is the single most central subsystem in core/.

Layers

graph TD
    User[User code action] --> SC[SparkContext]
    SC --> DAG[DAGScheduler]
    DAG --> TS[TaskSchedulerImpl]
    TS --> SB[SchedulerBackend]
    SB -->|launchTask RPC| EX[CoarseGrainedExecutorBackend]
    EX --> ETP[Executor task pool]

Layer	File	Responsibility
DAGScheduler	core/.../scheduler/DAGScheduler.scala	Build stages from RDD lineage, submit TaskSets, retry on failure.
TaskScheduler	core/.../scheduler/TaskSchedulerImpl.scala	Decide which executor runs which task; locality and fairness.
TaskSetManager	core/.../scheduler/TaskSetManager.scala	Per-stage state machine: speculation, exclusion, retries.
SchedulerBackend	core/.../scheduler/SchedulerBackend.scala	Pluggable cluster-manager interface.
CoarseGrainedSchedulerBackend	core/.../scheduler/cluster/CoarseGrainedSchedulerBackend.scala	Long-lived executor protocol used by all production backends.
ExecutorBackend	core/.../executor/CoarseGrainedExecutorBackend.scala	Per-executor RPC endpoint that runs Executor.

DAG construction

graph LR
    R[Final RDD] -->|narrow| R1[Narrow parent]
    R -->|shuffle| R2[Shuffle parent]
    R1 -->|narrow| R3
    R2 -->|narrow| R4

    subgraph "Stage 0 (ShuffleMapStage)"
        R3
        R4
    end

    subgraph "Stage 1 (ResultStage)"
        R
        R1
        R2
    end

DAGScheduler.handleJobSubmitted:

1. Walks the RDD's Dependency graph backwards.
2. Inserts a stage boundary at every ShuffleDependency.
3. Creates ShuffleMapStages for non-final stages and a single ResultStage for the final action.
4. Submits the leaf-most missing stages to the TaskScheduler.

Stages can be re-submitted on FetchFailed errors. The DAG scheduler is also where push-based shuffle's "merge finalize" step is coordinated and where AQE re-optimization hooks back into the scheduler when adaptive plans split or coalesce stages.

TaskScheduler

TaskSchedulerImpl orchestrates per-task placement. Per TaskSet:

• Maintains TaskSetManager (core/.../scheduler/TaskSetManager.scala) which tracks pending, running, and finished tasks.
• Honors locality preferences (PROCESS_LOCAL -> NODE_LOCAL -> RACK_LOCAL -> ANY).
• Implements speculation - launching a duplicate of a slow task on a different executor.
• Implements task-level exclusion via TaskSetExcludeList and node/executor exclusion via HealthTracker (core/.../scheduler/HealthTracker.scala).
• Pulls work from the Pool (core/.../scheduler/Pool.scala) which supports FIFO and FAIR scheduling between concurrent jobs.

SchedulerBackend implementations

Backend	Where
LocalSchedulerBackend	core/.../scheduler/local/
StandaloneSchedulerBackend	core/.../scheduler/cluster/StandaloneSchedulerBackend.scala
YarnClusterSchedulerBackend / YarnClientSchedulerBackend	resource-managers/yarn/.../scheduler/cluster/
KubernetesClusterSchedulerBackend	resource-managers/kubernetes/core/.../scheduler/cluster/k8s/

All except local extend CoarseGrainedSchedulerBackend, which speaks RegisterExecutor / LaunchTask / StatusUpdate over RPC.

Dynamic allocation

ExecutorAllocationManager (core/.../ExecutorAllocationManager.scala, the largest single file in the directory at ~49 KB) decides when to scale executors up or down based on:

• Pending tasks vs current parallelism.
• Idle executors past spark.dynamicAllocation.executorIdleTimeout.
• Active jobs vs spark.dynamicAllocation.minExecutors / spark.dynamicAllocation.maxExecutors.

It calls into ExecutorAllocationClient, which the scheduler backends implement.

Decommissioning

When an executor is told it is going away (preempted, spot-revoked, K8s node drain), it can gracefully migrate its blocks before exit. The flow is:

1. CoarseGrainedSchedulerBackend.decommissionExecutors marks the executor.
2. BlockManagerDecommissioner (core/.../storage/BlockManagerDecommissioner.scala) walks the blocks and pushes shuffle output to peers or to the FallbackStorage location.
3. Tasks running on the executor are killed cleanly and rescheduled.

Listener bus

LiveListenerBus (core/.../scheduler/LiveListenerBus.scala) is an asynchronous event bus that pumps SparkListenerEvents to listeners. It is the backbone of the Web UI, the metrics system, and many third-party plugins. AsyncEventQueue per-listener queues isolate slow listeners from fast ones.

Barrier execution

BarrierTaskContext (core/.../BarrierTaskContext.scala) and BarrierCoordinator add the ability for all tasks of a stage to start together, exchange messages, and exit together. This is the basis for distributed-training algorithms in MLlib.

Integration points

• The DAGScheduler is consulted by SQL execution (sql/core/.../execution/QueryExecution.scala) when an action is run.
• LiveListenerBus is the conduit for AQE plan updates - sql/core/.../execution/adaptive/AdaptiveSparkPlanExec.scala posts SparkListenerSQLAdaptiveExecutionUpdate events that the UI consumes.
• MapOutputTracker (core/.../MapOutputTracker.scala, ~75 KB) stores per-stage shuffle output locations and fan-out, used by reducers and AQE.

Entry points for modification

• Adding a SparkListener is the safest extension point. Implement the trait and register it via spark.extraListeners.
• Adding a new locality level: edit TaskLocality.scala and the placement logic in TaskSetManager.computeValidLocalityLevels.
• Adding a new exclusion criterion: edit HealthTracker.scala and TaskSetExcludeList.scala.
• Adding a new scheduler backend: implement SchedulerBackend and a matching ExternalClusterManager provider (core/.../scheduler/ExternalClusterManager.scala).