Executor memory

01 · What it is

One heap, four regions, two consumers that fight

An executor is a JVM process on a worker node that runs your tasks and holds cached data. Its heap is carved into four regions, and the one that runs out first sets the fate of your stage. The deep dive below breaks each down once; for now just hold the shape:

• Reserved (300 MiB) — a fixed slice carved off first for Spark internals.
• User memory (~40%) — your objects, UDF state, and metadata; not managed by the unified manager.
• Unified region M = spark.memory.fraction × (heap − 300 MiB), 0.6 by default — the shared pool for the two consumers that do the work: execution (shuffle / sort / agg / join buffers, transient) and storage (cached / persisted blocks + broadcast, long-lived).
• Storage floor R = spark.memory.storageFraction × M, 0.5 by default — the part of cache execution cannot evict.

analogy The executor heap is a shared commercial kitchen: a walled-off back office (reserved), your private pantry (user memory), and one big shared counter (region M). On that counter the active cooking workspace (execution) and the shelf of prepped ingredients (storage/cache) share the surface — the cook can shove ingredients aside, but the shelf can never push the cook off.

02 · Why it matters

Spill and OOM are the two tickets you'll actually get

• They are the most common "slow / dead job" tickets — and both are memory-region problems you can only diagnose once you know which region (execution vs storage) was under pressure.
• Spill is a silent tax. When execution memory is exhausted, Spark spills to local disk and the job still finishes — just much slower (disk I/O + extra GC). Many "slow Spark jobs" are really un-noticed spill.
• Executor OOM kills the task when even spilling can't keep a partition within the container limit — usually a skewed partition, cache pinned in R, or oversized Python workers.
• A classic interview probe: "What's the difference between spill and OOM, and why can't storage evict execution?" The how is the next section.

03 · How it works — deep dive

The regions, the asymmetry, and the two ways it fails

Everything an executor-memory problem comes down to is where the data sits while a task runs. So the useful way to understand it is: (1) what the regions are and how they're sized, (2) the borrow-and-evict rule that governs the shared pool, and (3) the two failure modes — spill, then OOM — plus the memory the heap diagram doesn't show. Each idea is explained here once; later sections show it, verify it, and help you decide, but don't re-explain it.

3.1 · The four regions, sized

How the heap is partitioned, in order, on a default spark.executor.memory = 1g executor (usable = heap − 300 MiB ≈ 724 MiB):

use case That ~434 MiB of M is all that's shared between every shuffle buffer and every cached block on the executor — which is why a heavy shuffle and a big cache in the same stage fight over the same few hundred MiB, and someone spills.

3.2 · Two consumers, two lifecycles

• Execution memory — buffers for shuffle, sort, aggregation, join (the hash map of a hash-aggregation, the sort buffer of a sort-merge join). Transient: grabbed by a task, freed when it finishes. When short → spill.
• Storage memory — cached / persisted blocks + broadcast. Long-lived: stays until evicted or unpersist()-ed. Over-allocated → starves execution.

analogy Execution is scratch paper you scribble on and toss mid-task; storage is a reference book you keep on the desk all session.

3.3 · The unified model — borrow & the evict asymmetry

M is unified, not a static 60/40 split. Three rules govern it:

• Borrow freely. Doc: "When no execution memory is used, storage can acquire all the available memory and vice versa." Nothing cached → a shuffle can use all of M; nothing shuffling → cache can fill all of M.
• Execution may evict Storage — down to R. A task needing execution memory evicts cached blocks above R. Blocks inside R (= 0.5 × M) are immune.
• Storage may NOT evict Execution. If cache wants more but execution holds it, cache just doesn't get it — the new block spills to disk (per its storage level) or isn't cached. Execution is never taken from a running task.

use case You cache() a 400 MiB DataFrame then aggregate it; the aggregation evicts cache blocks above R, so the cache ends up smaller than expected and the next read partly recomputes — the asymmetry biting a real job.

3.4 · Spill — the warning sign before death

• What it is. When execution memory is exhausted during a shuffle/sort/agg/join, Spark spills the in-memory structures to local disk. Doc: "When data does not fit in memory Spark will spill these tables to disk, incurring the additional overhead of disk I/O and increased garbage collection."
• What it means. The job still finishes — graceful degradation, not a crash — but pays disk I/O + more GC, so a spilling stage can be several times slower.
• How to see it. Spark UI → Stages shows Spill (Memory) and Spill (Disk) per task/stage. Any non-zero spill is a flag to investigate.

analogy Spill is overflow parking: the lot (execution memory) is full, so cars (data) park in the muddy field down the road (disk). Everyone still parks — it's just slower to walk back.

3.5 · Executor OOM — when spilling isn't enough

OOM (java.lang.OutOfMemoryError) happens when even spilling can't keep a partition within the container limit. The four causes, and the fix that actually works for each:

use case A groupBy where 90% of rows share one value → one task builds a hash aggregation for 90% of the data → spills, spills, then exceeds the container and the executor dies. Fix the skew, don't just add RAM.

3.6 · The memory outside the bar — off-heap and Python workers

• Off-heap. Memory outside the JVM heap (Tungsten/Unsafe), so it's not garbage-collected. spark.memory.offHeap.enabled = false (default), spark.memory.offHeap.size = 0 (must be > 0 when enabled). When on, the docs advise you "shrink your JVM heap size accordingly." Mainly a win when GC is the bottleneck (Lesson 10).
• Python-worker memory (the hidden PySpark OOM). Python UDFs, mapInPandas, Pandas/Arrow UDFs, and RDD map in Python run in separate Python worker processes outside the JVM heap — they do not use region M. spark.executor.pyspark.memory is Not set by default; when unset, Spark doesn't limit them and they count against the container's overhead. Overhead = executor.memory × 0.10 (factor; 0.40 for K8s non-JVM), floored at 384m — covers off-heap, shuffle/network buffers, and the Python workers.

use case A Pandas UDF that materializes a big batch per group blows the container while the JVM heap looks fine. The cure is more overhead (or a smaller batch / native function), not a bigger spark.executor.memory.

Figure A · borrow, evict, spill, OOM

Watch the shared region fill

Drag execution demand and storage (cache) demand and watch execution borrow free space, then evict cache down to the R floor when it needs more — while cache can never push execution back. Push total demand past M and a hatched spill segment appears; push it well past the heap and the executor OOMs. This figure isolates one thing: the regions and their thresholds. (Where it runs across the cluster is Figure B.)

04 · The code — and how to verify it

Four moves, each with its proof

Every technique below is paired with how to prove it worked — the plan node or the Spark-UI signal. Apply, then verify; never assume.

Inspect the unified-memory configs

python# The four numbers that define region M. Read them before tuning anything.
print(spark.conf.get("spark.executor.memory"))          # default "1g" — the JVM heap
print(spark.conf.get("spark.memory.fraction"))          # default "0.6" — M = 0.6 * (heap - 300 MiB)
print(spark.conf.get("spark.memory.storageFraction"))   # default "0.5" — R = 0.5 * M (eviction-immune cache)

# VERIFY (do the arithmetic): on a 1g executor,
#   usable = 1024 - 300 = 724 MiB ; M = 0.6 * 724 ≈ 434 MiB ; R = 0.5 * 434 ≈ 217 MiB
# That ~434 MiB is shared by EVERY shuffle buffer AND every cached block on the executor.

See (and cure) a spill

pythonfrom pyspark.sql import functions as F

# CREATE the condition: a wide aggregation over many groups stresses EXECUTION memory.
big = spark.range(0, 200_000_000).withColumn("k", (F.rand() * 50_000_000).cast("long"))
agg = big.groupBy("k").count()

# VERIFY the spilling operator is in the plan:
agg.explain(mode="formatted")
#   look for: HashAggregate(keys=[k], ...)  +  Exchange hashpartitioning(k, 200)
#   HashAggregate is the execution-memory consumer that spills when M is exhausted.

# Run the job, then READ THE UI:
agg.write.format("noop").mode("overwrite").save()
# Spark UI → Stages → aggregation stage → columns "Spill (Memory)" / "Spill (Disk)".
# Non-zero spill = execution memory ran out. Cure: more, smaller partitions:
spark.conf.set("spark.sql.shuffle.partitions", 800)   # smaller slices → less per-task memory
# Re-run and confirm the spill columns drop toward zero. (Reset the conf afterwards.)

Watch execution evict storage (the asymmetry)

pythonfrom pyspark import StorageLevel

# Cache a DataFrame, then run heavy EXECUTION work on it in the same stage.
cached = big.persist(StorageLevel.MEMORY_AND_DISK)   # DataFrame default; spills cache to disk
cached.count()                                        # materialize (cache is lazy until an action)

heavy = cached.groupBy("k").count()                   # execution now competes with the cache
heavy.write.format("noop").mode("overwrite").save()

# VERIFY — Spark UI → Storage tab: the cached DataFrame may show < 100% "Fraction Cached".
# Execution EVICTED cache blocks above R to make room. Storage could not evict execution.
cached.unpersist()                                    # always release cache when done (frees R)

Budget the memory outside the heap (off-heap & Python)

python# Off-heap lives OUTSIDE the JVM heap → no GC. Size it explicitly and shrink the heap
# (per the tuning docs) so you don't double-count RAM. Set at CLUSTER creation in prod.
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size", str(2 * 1024 * 1024 * 1024))  # 2 GiB in bytes (>0)

# Python UDFs run OUTSIDE the heap, against the container's OVERHEAD (not runtime-settable):
print(spark.conf.get("spark.executor.pyspark.memory", "<not set>"))   # → "<not set>" by default
#   spark.executor.memoryOverhead = "2g"   # a floor for Python-heavy work (cluster config)

# VERIFY — Spark UI → Executors tab: "Off Heap Memory Used" becomes non-zero and GC Time drops;
# container OOM kills stop after raising OVERHEAD — bumping spark.executor.memory would NOT help.

Figure B · the task, filling one executor

See it run across the cluster

This figure isolates a different thing from Figure A: not the thresholds, but the order of events over time. Watch one aggregation on Executor A (two more shown for context): execution grows in M, evicts cached blocks down to the R floor, spills to local disk when it exceeds M, then a skewed partition outruns spill and the executor OOMs. Use Prev / Play / Next.

05 · Comparison at a glance

Execution vs storage — the two consumers, side by side

Read this by the consumer you're worried about — it answers "how does this region behave under pressure?" at a glance. (The why is §3; this is the contrast.)

06 · Edge cases & limitations

Where it gets subtle

New nuance an interviewer probes — not a re-run of §3:

• Skew AQE can't fully fix. A key so dominant that even split sub-partitions exceed memory; fall back to salting (Lesson 08) or a different key.
• Cache pinned in R. Heavily-cached executors look "fine on heap" but OOM under a shuffle — because R is immune to eviction. The cure is less cache, not more memory.
• Broadcast in storage. A broadcast relation consumes storage memory on every executor; a near-threshold broadcast (Lesson 02) pressures M cluster-wide.

Hard limits worth stating plainly:

• The unified model manages only the JVM heap regions (M + user) — not Python workers, off-heap (unless enabled), or the container limit. Those are separate budgets you must size yourself.
• spark.memory.fraction / storageFraction are cluster/JVM-level. Changing them mid-session does not resize a running executor — set them at cluster creation.
• Spill makes a job slow, not safe forever. A large enough skewed partition still OOMs no matter how much it spills.
• OSS vs Databricks. The unified-memory math (0.6 / 0.5 / 300 MiB / overhead 0.10) is identical; Databricks may layer its own overhead / Photon accounting on top — verify sizing in the Azure Databricks compute docs.

07 · Common mistakes

Gotchas — the mistake and its one-line fix

08 · References

Sources

• Apache Spark — Tuning Guide (memory management, unified model, spill, GC)
• Apache Spark — Configuration (executor.memory, memory.fraction, storageFraction, offHeap.*, memoryOverhead, pyspark.memory)
• Apache Spark — Cluster overview (executors, what runs where)
• Azure Databricks — Compute configuration (cluster / executor sizing)