Spark Core · Module 1 · Deep dive

Inside the executors

How tasks fill the cluster, how executor memory is really divided, and the two features that change executors at runtime: dynamic allocation and speculative execution. Start with the overview if you need the basics.

Slots & waves Memory in detail Dynamic allocation Speculation Deploy modes Check yourself

Slots, waves and stage duration

Each executor core is a slot that runs one task at a time. With 10 executors × 4 cores you have 40 slots, so a stage of 1,000 tasks runs in about 1,000 ÷ 40 = 25 waves. Picture 40 checkout counters serving 1,000 customers.

At about 20 s per task, 25 waves take roughly 500 s, or 8–9 minutes. In practice a slot picks up the next task as soon as it's free, so the waves blur together, but the total is the same.

What actually shortens a stage

More slots. 80 slots → about 13 waves → roughly half the time. This is the main lever.
No half-empty last wave. 1,010 tasks need 26 waves, and the last runs 10 tasks while 30 cores sit idle. Keep task counts near a multiple of total cores.
Merging only tiny tasks. Fewer, larger partitions don't speed up a stage: the work is the same, each task just runs longer. It helps only when tasks take around 100 ms and scheduling overhead dominates.
Fixing skew. A stage ends when its slowest task ends. One 15-minute task makes the stage take at least 15 minutes, however many slots you add.

Executor memory in detail

The same 16 GB, 4-core executor as on the overview, drawn as nested regions. Each box is a separate budget; running out of a different box gives a different error.

The dashed line between storage and execution is a soft boundary. Execution can push it left by evicting cached blocks, but only down to the protected storage half (spark.memory.storageFraction = 0.5). Cache can push it right only into execution memory that's free.

Try your own executor size

spark.executor.memory (GB) spark.executor.cores Overhead factor

Defaults used: 300 MB reserved, spark.memory.fraction = 0.6, spark.memory.storageFraction = 0.5, overhead = max(384 MB, factor × heap).

Why cores matter as much as memory

The unified pool is shared by every task running at once. For the same 16 GB executor:

Cores per executor	Tasks at once	Per task, no cache	Per task, cache at its half
2	2	4.7 GB	2.4 GB
4	4	2.4 GB	1.2 GB
8	8	1.2 GB	0.6 GB

Halving cores doubles the memory each task gets, without buying any RAM. The trade-off is less parallelism per executor.

How the pool behaves under pressure

Execution needs more: it takes free storage memory, then evicts cached blocks down to the protected half.
Storage needs more: it can only use free execution memory. Otherwise new blocks are dropped (MEMORY_ONLY) or written to disk (MEMORY_AND_DISK).
Execution runs out: sorts, joins and aggregations spill sorted runs to local disk and merge them later. Slow, but the task survives.
Something that can't spill runs out: one huge record, an unbounded collect_list, a hash table that must stay in memory, or objects in user memory. That's an OutOfMemoryError.

Memory that lives outside the heap

Overhead (spark.executor.memoryOverhead, or spark.executor.memoryOverheadFactor since Spark 3.3): JVM metaspace and thread stacks, Netty network buffers used by shuffle, native compression libraries, and Python worker processes.
PySpark memory (spark.executor.pyspark.memory, unset by default): when set, caps each executor's Python workers and is added to the container request. When unset, Python memory must fit inside overhead.
Off-heap execution memory (spark.memory.offHeap.enabled + spark.memory.offHeap.size): lets Tungsten keep data outside the JVM heap to cut GC pressure. Since Spark 3.0 it's added to the container size on YARN and Kubernetes.

Container size = heap + overhead + PySpark memory (if set) + off-heap size (if enabled). The cluster manager kills the container when the process goes over that total.

Dynamic allocation

The problem it solves. With static allocation (--num-executors 50) a job holds all 50 executors for its whole life, even during a stage with 10 tasks or while the driver sits idle between jobs. On a shared YARN or Kubernetes cluster that blocks other teams and wastes money.

What it does. Spark grows and shrinks the number of executors based on how many tasks are waiting.

How it decides

Direction	Trigger	Detail
Scale up	Tasks pending longer than `schedulerBacklogTimeout` (1 s)	Requests double each round (1, 2, 4, 8 …) while the backlog lasts (`sustainedSchedulerBacklogTimeout`, 1 s), up to `maxExecutors`
Scale down	An executor ran no tasks for `executorIdleTimeout` (60 s)	Never below `minExecutors`. Executors holding cached data use `cachedExecutorIdleTimeout`, which is infinite by default, so they're kept

spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=2
spark.dynamicAllocation.initialExecutors=2
spark.dynamicAllocation.maxExecutors=50          # always set a cap on shared clusters
spark.dynamicAllocation.executorIdleTimeout=60s
spark.dynamicAllocation.shuffleTracking.enabled=true   # or spark.shuffle.service.enabled=true

The catch: shuffle files

Executors write shuffle output to their own local disk, and the next stage reads it from there. If Spark removed an idle executor, that data would be lost and the parent stage would have to rerun. So dynamic allocation needs one of these:

External shuffle service (classic on YARN): a long-running process on each node serves shuffle files, so executors can leave safely.
Shuffle tracking (Spark 3.0+, common on Kubernetes, where there's no shuffle service): Spark keeps alive any executor that still holds shuffle data a later stage needs.
Graceful decommissioning (Spark 3.1+, spark.decommission.enabled): before an executor leaves, for example when a spot node is reclaimed, its shuffle and cache blocks are migrated to other executors.

When to use it

Good fit	Poor fit
Shared clusters; notebooks and ad-hoc work; jobs whose stages vary a lot in size; cost control	Latency-sensitive jobs (new executors take seconds to minutes to start, longer if new nodes are needed); Structured Streaming, which wants steady capacity; jobs that cache heavily (cached executors never scale down)

On Databricks you set cluster autoscaling (min/max workers) instead. It adds and removes worker nodes, the platform-level version of the same idea.

Common trap

Dynamic allocation with no maxExecutors on a shared cluster. One job with a large backlog can request every free container and starve everyone else. Always set a cap.

Say it in an interview

"Dynamic allocation requests executors when tasks are backlogged for more than a second, doubling each round up to a max, and releases them after 60 seconds idle. It needs an external shuffle service or shuffle tracking so removing an executor doesn't lose shuffle files. I use it on shared or bursty workloads with a sensible max, and avoid it for latency-critical or streaming jobs."

Speculative execution

The problem it solves. Sometimes one task is slow for reasons unrelated to its data: a failing disk, an overloaded node, a noisy neighbour, a slow network link. The stage can't finish until that task does.

What it does. When enabled (spark.speculation=true, off by default), Spark watches running tasks and launches a duplicate copy of a straggler on a different executor. Whichever copy finishes first wins; the other is killed.

When a task counts as a straggler

Spark checks every spark.speculation.interval (100 ms). A running task gets a copy when both are true:

At least spark.speculation.quantile (75%) of the stage's tasks have finished, and
the task has been running longer than spark.speculation.multiplier (1.5) × the median duration of finished tasks.

spark.speculation=true
spark.speculation.multiplier=1.5       # raise to 2–3 to be less aggressive
spark.speculation.quantile=0.75
spark.speculation.minTaskRuntime=100ms # don't speculate on very short tasks

Classic trick question: does speculation fix skew?

No. A skewed task is slow because its partition holds far more data. The copy gets the same partition, so it's just as slow, and you've wasted a core. Speculation helps only when the machine is the problem, not the data. Tell them apart in the Spark UI: a skewed task has a much larger Shuffle Read Size than its peers; a bad-node task has normal input but a long duration, often with others on the same host.

Risks

Duplicate side effects. If a task writes to an external system (a JDBC insert in foreachPartition, an API call, sending messages), both copies may write. File and table writes are protected by Spark's output committer; custom writes aren't. Make them idempotent or keep speculation off.
Wasted capacity. Every copy takes a slot that real work could use. On a busy cluster, aggressive settings slow everything down.
Hidden root cause. Speculation can mask a sick node for weeks. Watch for the same host repeatedly producing stragglers and fix or exclude it.

Say it in an interview

"Speculation re-launches a task that's running much slower than its peers once most of the stage is done: by default 1.5 times the median after 75% of tasks finish. Whichever copy finishes first wins. It helps with bad nodes, not skew, because the copy gets the same data. I'd avoid it when tasks write to non-idempotent external systems."

Client vs cluster deploy mode

	Client mode	Cluster mode
Where the driver runs	On the machine that ran spark-submit (or the notebook host)	Inside the cluster, on a node the cluster manager picks
Good for	Interactive work, notebooks, debugging	Scheduled production jobs
Risk	If the laptop or edge node disconnects, the job dies; network latency to executors	Logs live in the cluster, so it's harder to watch interactively

On Databricks you don't choose this: each cluster has a driver node and worker nodes, with one executor per worker. You tune by choosing instance types and worker counts.

Check yourself

Your 64 GB executors have 16 cores each and spill constantly. What would you change first, and why?

Fewer cores per executor (for example 4–5), with more executors. Each task currently gets roughly 64 × 0.6 ÷ 16 ≈ 2.3 GB of pool; at 4 cores it's about 9 GB. Very large heaps also mean longer GC pauses.

Dynamic allocation is on, but executors never scale down. Why?

Most likely they hold cached data (cachedExecutorIdleTimeout is infinite by default), or shuffle tracking is keeping them because a later stage still needs their shuffle files. Unpersist cached data you no longer need, or set a finite cached idle timeout.

Why does dynamic allocation need an external shuffle service or shuffle tracking?

Shuffle output lives on executors' local disks. Removing an executor without either feature would lose shuffle files that the next stage needs, forcing recomputation of the parent stage.

One task in a stage takes 10× the median. Speculation or salting?

Check its Shuffle Read Size first. If it's 10× the median too, it's skew: salting, AQE skew join or broadcast. If its input is normal, it's the node: speculation helps, and you should look at that host.