Locality and Scheduling Delay

Concepts covered: Spark Execution Model

When the DAGScheduler builds a task set it also records, for each task, where its input already lives: for files on the cluster's own disks, the nodes holding that file's blocks. The TaskScheduler uses those preferred locations when it hands out cores. A task placed next to its data skips a network read, and the price of insisting on it is a free core elsewhere left waiting.

Level	Where the task's input is	Typical cause
PROCESS_LOCAL	In the memory of the executor running the task	Data that executor computed earlier and kept in memory
NODE_LOCAL	On the same machine, on local disk or in another executor	File blocks stored on the cluster's own disks
NO_PREF	Equally far from every executor	Rows from a source with no location, such as a remote database
RACK_LOCAL	On another machine in the same rack	File blocks stored on a neighbouring node
ANY	Anywhere else on the network	No better core appeared within the wait

Data in a remote object store has no copy on any executor's machine, so its tasks run at NO_PREF or ANY and locality hardly matters.

Delay scheduling

When a core frees up on a machine holding none of a pending task's data, the TaskScheduler can run a task there now at a worse level, or hold the task briefly in case a core frees up next to its data. Spark holds it. The wait per level is spark.locality.wait, 3s by default, with overrides in spark.locality.wait.process, spark.locality.wait.node and spark.locality.wait.rack. The timer belongs to the task set: after 3s with no launch at its current level, the set relaxes one level, and it keeps relaxing until ANY.

Here a core on node K frees up. K sits in a rack holding none of the stage's input, and no core frees up next to the data meanwhile:

Time	Free core on	Best level allowed	Decision
0 s	node K	NODE_LOCAL	Hold: no pending task has data on K
3 s	node K	RACK_LOCAL	Hold: K's rack holds none of the data
6 s	node K	ANY	Launch the next pending task on K

Short stages that lose more to waiting than they save

A job runs 40 short stages, each reading files whose blocks sit on 10 of the cluster's 40 nodes, with tasks of about 1.5 seconds. At the start of every stage, the 10 nodes holding data fill with local tasks, while offers from the other 30 are turned down for up to 3 seconds at NODE_LOCAL and up to 3 more at RACK_LOCAL. That is up to 6 seconds of idle cores on three quarters of the cluster, 40 times over, spent to save network reads of about a second each. Setting the wait to 0s or 1s lets those executors start at once.

Lower spark.locality.wait when

Tasks run a few seconds or less, so 3 s of waiting rivals the task itself
Input sits in an object store or remote service, so no executor is local anyway
The Event Timeline shows idle executors at each stage's start while tasks are pending

Keep the default when

Tasks read hundreds of megabytes each from disks inside the cluster
The network between nodes is slower than local disk, or already saturated
The Locality Level Summary shows mostly NODE_LOCAL and executors stay busy

Two queries run 1-second tasks. Query A reads from a remote object store; query B reads files stored on 8 of the cluster's 40 nodes. Only B's stage starts with executors idle while tasks are pending. Why?

Disks inside the cluster read more slowly, so B's executors wait on their reads.
The driver places tasks with preferred hosts more slowly, so B's tasks launch late.
B's tasks prefer the 8 nodes holding their blocks, so delay scheduling holds them; A's have none.
Delay scheduling holds both queries' tasks, but A's wait is hidden at the NO_PREF level.

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("short-stage-report")
    .config("spark.locality.wait", "1s")
    .getOrCreate()
)

A stage of 600 tasks, each running about 2 seconds, reads files stored on the cluster's own disks. Half the executors sit idle for its first 6 seconds while tasks are pending, and the Locality Level Summary shows 300 NODE_LOCAL and 300 ANY. Why the idle start?

The driver was slow to launch 600 tasks, so half the executors waited for the scheduler to reach them.
Delay scheduling held tasks for data-local cores, waiting 3 s per level before running them at ANY elsewhere.
Half the tasks were speculative copies, which wait 6 s so that their originals get a chance to finish.
The idle executors were reading input over the network, which the timeline draws as gaps before each task.

About this lesson

This section is part of the Execution Model: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.