Author(s): chakshu_salgotra

Originally published on Towards AI.

How Spark Manages Memory — The Unified Memory Model, Spills, and AQE

Where every gigabyte in an executor actually goes, and why your job dies at 85% completion

Part I | Part II | Part III | Part IV | Part V

The OOM That Only Struck at 85%

A daily aggregation over 2.1TB of event data ran clean for weeks. Then, without a code change, it started dying — always at the same place: 85% through the final stage, with Container killed by YARN for exceeding memory limits. 9.4 GB of 9 GB physical memory used. Restart it, and it died again at 85%. Bump spark.executor.memory from 8g to 12g, and it died at 85% — just a few minutes later.

The team did what most teams do: threw memory at it. executor.memory went to 16g, then 20g. The job survived at 20g, at roughly 2.4x the cluster cost, and nobody understood why. The real problem wasn't the total heap — it was how Spark had divided that heap, and the fact that a single skewed reduce partition was demanding more execution memory than the region was allowed to grow into, while gigabytes of cached storage memory sat pinned and untouchable right next to it.

The failure wasn’t “not enough memory.” It was a boundary dispute inside the executor. So: how does Spark actually partition an executor’s memory, when can one region borrow from another, why does data spill, and how does Adaptive Query Execution quietly defuse the exact skew that was killing this job?

The Executor Is Not One Pool of Memory

The first model to discard: an executor’s memory is not a single heap that Spark fills until it’s full. A running executor’s JVM heap is carved into distinct regions, each with its own rules, and an OOM almost always means one specific region overflowed — not that the container ran dry.

Since Spark 1.6, allocation is governed by the Unified Memory Manager (UnifiedMemoryManager). It splits the usable heap into four regions, computed from the JVM's Runtime.maxMemory:

• Reserved Memory — a hardcoded 300MB floor, set aside for Spark’s own internals. You cannot use it, and Spark refuses to start if the heap can’t clear it comfortably (it requires at least 1.5 × 300MB).
• User Memory — everything outside spark.memory.fraction. This holds your data structures: the objects inside a UDF, arrays you build in a mapPartitions, RDD lineage metadata. Spark does not track or manage it; overrun it and you OOM directly.
• Unified Region (Execution + Storage) — the large managed pool governed by spark.memory.fraction (default 0.6), where the real work happens.

The Executor Heap, Region by Region

+---------------------------+ <- Runtime.maxMemory
| Reserved (300MB) |
+---------------------------+
| |
| User Memory (~40%) | UDF objects,
| (1 - memory.fraction) | user structures
| |
+---------------------------+ <- memory.fraction (0.6)
| |
| Storage Memory | cached blocks,
| (borrowable) | broadcasts
|- - - - - - - - - - - - - -| <- storageFraction (0.5)
| |
| Execution Memory | shuffles, joins,
| (borrowable) | sorts, aggregations
| |
+---------------------------+

The spark.memory.fraction × heap slice is the unified region, and inside it sits the single most important idea in Spark memory management.

Execution vs. Storage: The Borrowing Contract

The unified region is shared between two competing consumers, and the rules of that competition explain most OOMs and most spills.

• Execution Memory is transient, working memory for the machinery of a query: shuffle buffers, hash tables for joins and aggregations, and sort buffers. It is allocated per-task and released the moment a task finishes.
• Storage Memory is for cached/persisted RDDs and DataFrames, broadcast variables, and the blocks pinned by .cache() / .persist().

spark.memory.storageFraction (default 0.5) draws a soft line down the middle of the unified region. "Soft" is the key word — the entire point of the unified model (versus the old static split) is that either side can borrow the other's unused space:

• Execution can evict Storage. If a task needs execution memory and storage is holding cached blocks in the borrowable zone, Spark will evict those cached blocks (spilling them to disk or dropping them, per the block’s StorageLevel) to hand the space to execution. Execution always wins.
• Storage can borrow from Execution — but it cannot forcibly evict execution memory, because execution memory in flight can’t be spilled without failing a task. Storage can only use execution’s idle space, and must give it back on demand.

The asymmetry is deliberate: execution memory is unspillable-in-the-moment and always takes priority; storage is evictable and yields. This is exactly what bit the incident. Cached blocks below the storageFraction line are protected — execution cannot evict them. The skewed task needed more execution memory than the borrowable zone could supply, because a large .cache() had parked protected storage blocks it wasn't allowed to touch.

Why Data Spills — and Why That’s a Feature

When a task needs more execution memory than it can acquire (even after evicting borrowable storage), Spark does not immediately OOM. It spills: it sorts the in-memory contents, writes them to local disk as a sorted spill file, frees the memory, and continues. At the end, it merges the spill files with whatever remained in memory.

Spilling is the safety valve that lets a 2TB shuffle complete inside a few GB of execution memory. But it is not free:

• Every spill is a serialize → sort → write → re-read → merge → deserialize round trip to disk.
• Spill (memory) in the Spark UI is the in-memory size; Spill (disk) is the smaller serialized on-disk size. A stage where spill dwarfs input is telling you execution memory is starved.

The OOM at 85% happened because the skewed partition’s hash-aggregation buffer couldn’t spill fast enough relative to how quickly one monster partition grew — a single task tried to build one enormous in-memory map, and the container’s hard physical limit (heap + overhead) was breached before spill could rescue it. Which brings in the region everyone forgets.

Off-Heap and the Overhead Trap

spark.executor.memory sizes the JVM heap — but the container the scheduler kills is bounded by heap plus spark.executor.memoryOverhead (default max(384MB, 0.1 × executor.memory)). Overhead covers off-heap shuffle buffers, network buffers, and Tungsten's off-heap allocations. Container killed by YARN almost always means overhead, not heap, was exceeded — raising executor.memory alone can make it worse by shrinking the overhead's relative share.

Tungsten can also move execution memory off-heap entirely (spark.memory.offHeap.enabled, sized by spark.memory.offHeap.size), storing data in compact binary form outside the JVM heap. This dodges garbage-collection pauses on the hottest data path — the GC pressure from millions of short-lived aggregation objects is itself a common cause of executor stalls.

Where AQE Changes the Memory Equation

Here’s the part most memory tuning guides miss: the cheapest way to fix a memory problem is often to not create the skew in the first place. Adaptive Query Execution (spark.sql.adaptive.enabled, on by default since 3.2) re-plans the query at runtime using real shuffle statistics — and three of its optimizations are memory interventions in disguise.

• Skew-join splitting (spark.sql.adaptive.skewJoin.enabled) — AQE detects a partition far larger than the median (per skewJoin.skewedPartitionFactor and skewedPartitionThresholdInBytes) and splits it into sub-partitions, each handled by its own task. The one fat reduce partition that demanded impossible execution memory becomes several normal ones. This is the automatic version of manual key salting.
• Coalescing shuffle partitions (spark.sql.adaptive.coalescePartitions.enabled) — after a shuffle, AQE merges the many tiny post-shuffle partitions into fewer right-sized ones (target advisoryPartitionSizeInBytes, ~64MB). Fewer, appropriately sized tasks mean less scheduler overhead and steadier memory use per task.
• Dynamic join switching — when a side turns out smaller than estimated, AQE promotes a SortMergeJoin (which shuffles and sorts both sides, heavy on execution memory) to a BroadcastHashJoin (no reduce side, no shuffle-sort buffers).

You can read the runtime decision straight from the plan — the final plan is wrapped as AdaptiveSparkPlan isFinalPlan=true, and skew handling shows up as extra partitions on the exchange.

Code: Reshaping the Memory Footprint

The naive job cached aggressively and let one skewed key build a single monster aggregation buffer. The engine-aware fix is to stop protecting storage that execution needs, and let AQE split the skew.

# NAIVE: MEMORY_ONLY cache pins protected storage blocks that execution
# can't evict, then a skewed aggregation starves for execution memory -> OOM.
df.persist() # defaults to MEMORY_AND_DISK, blocks sit in the protected zone
result = df.groupBy("user_id").agg(F.sum("amount"))

# OPTIMIZED: let AQE split the skewed partition, and cache to disk-backed
# level so execution can reclaim memory instead of OOM-ing.
from pyspark import StorageLevel

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")# DISK_ONLY frees the borrowable storage zone for execution to grow into.
df.persist(StorageLevel.DISK_ONLY)
result = df.groupBy("user_id").agg(F.sum("amount"))

Tune the memory split explicitly when the workload is execution-heavy (big shuffles, little caching) — shrink storage’s protected share so execution has more room before it spills:

# Execution-heavy job: give execution the whole unified region by default.
# storageFraction 0.3 shrinks the *protected* storage floor, so execution
# can borrow (and evict down to) far more before spilling to disk.
spark.conf.set("spark.memory.fraction", "0.6") # unified region size
spark.conf.set("spark.memory.storageFraction", "0.3") # protected storage floor

# Move the hot aggregation path off-heap to dodge GC pauses entirely.
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size", "4g")

For containers dying with YARN killed, fix overhead, not heap — the mistake the incident kept making:

# NAIVE: bump total heap and hope. Shrinks overhead's relative share.
# spark.executor.memory = 20g (overhead still ~2g -> still killed)
# OPTIMIZED: size overhead for off-heap shuffle/network buffers explicitly.
# spark.executor.memory = 12g
# spark.executor.memoryOverhead = 4g # was the real ceiling all along

Confirm the skew fix landed by reading the runtime plan — AQE annotates the split:

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
 *(5) HashAggregate(keys=[user_id])
 +- AQEShuffleRead coalesced + skewed <- AQE split the fat partition
 +- ShuffleQueryStage

Tuning Memory in Production

• Read the region, not the total. An OOM is one region overflowing. Check whether spill is execution-side (raise memory.fraction or lower storageFraction) or whether you're OOM-ing in User Memory inside a UDF (no config fixes that — shrink your per-task objects).
• Container killed by YARN means overhead. Raise spark.executor.memoryOverhead before spark.executor.memory. Off-heap shuffle and network buffers live there, not in the heap.
• Cache deliberately. MEMORY_ONLY blocks in the protected zone are memory execution can't reclaim. Prefer MEMORY_AND_DISK or DISK_ONLY on execution-heavy jobs, and un-persist() the moment a dataset is done.
• Let AQE fight skew first. Skew-join splitting and partition coalescing eliminate the fat-partition OOMs that manual heap-bumping only postpones.
• Watch GC. Millions of short-lived aggregation objects thrash the JVM. Off-heap execution memory (spark.memory.offHeap.enabled) moves the hot path out of GC's reach.

Result on the incident: enabling AQE skew-join splitting broke the monster reduce partition into eight, switching the aggressive MEMORY_ONLY cache to DISK_ONLY freed the protected storage zone for execution, and right-sizing memoryOverhead to 4g fixed the actual container ceiling. The job ran green at 12g executors — below the original 8g-that-failed cluster cost when accounting for the abandoned 20g workaround — and the 85% cliff disappeared.

Takeaways

• An executor is four regions, not one pool. Reserved (300MB), User Memory (outside memory.fraction), and the unified Execution/Storage region inside it. An OOM is always one region overflowing.
• Execution beats Storage. Execution memory can evict borrowable cached blocks and always wins; storage can only borrow execution’s idle space and must yield it. Blocks below storageFraction are protected and cannot be evicted — a common self-inflicted starvation.
• Spilling is the safety valve, overhead is the hidden ceiling. Spills trade disk I/O for survival; Container killed by YARN almost always means memoryOverhead, not heap, was breached.
• AQE is memory management in disguise. Skew-join splitting, partition coalescing, and dynamic broadcast switching remove the skew and buffer pressure that manual heap-tuning only delays.

Stop feeding the executor more heap. Find out which region is on fire.

Glossary

The specialized terms above, in roughly the order they appear.

• Unified Memory Manager (UnifiedMemoryManager) — the component (since Spark 1.6) that dynamically shares heap between execution and storage, replacing the old static split.
• Reserved Memory — a hardcoded 300MB slice reserved for Spark internals; unusable by your workload.
• User Memory — the heap outside spark.memory.fraction (~40% by default), holding user data structures and UDF objects; untracked by Spark, so overruns OOM directly.
• Unified region — the spark.memory.fraction slice (default 0.6) shared between execution and storage.
• spark.memory.fraction — the fraction of usable heap given to the managed unified region (default 0.6).
• Execution Memory — transient working memory for shuffles, joins, aggregations, and sorts; allocated per task, always takes priority, can evict storage.
• Storage Memory — memory for cached/persisted blocks and broadcasts; evictable, yields to execution.
• spark.memory.storageFraction — the soft, protected floor for storage inside the unified region (default 0.5); blocks below it cannot be evicted by execution.
• Borrowing / eviction — execution may evict borrowable storage blocks to reclaim space; storage may only use execution’s idle memory and must return it.
• Spill — sorting and writing in-memory data to local disk when execution memory is exhausted, then merging on read; the safety valve that prevents OOM.
• Spill (memory) / Spill (disk) — the in-memory size versus the smaller serialized on-disk size of spilled data, as shown in the Spark UI.
• spark.executor.memoryOverhead — off-heap memory for shuffle, network, and Tungsten buffers (default max(384MB, 0.1 × executor.memory)); the real ceiling behind Container killed by YARN.
• Off-heap memory — memory Spark manages directly outside the JVM heap (spark.memory.offHeap.enabled / .size), avoiding GC pauses.
• Tungsten — Spark’s execution backend that stores data in compact binary form (often off-heap) and generates optimized code to cut CPU and GC overhead.
• StorageLevel — the caching policy for a persisted dataset (MEMORY_ONLY, MEMORY_AND_DISK, DISK_ONLY, …) that decides where blocks live and whether they can spill.
• AQE (Adaptive Query Execution) — runtime re-optimization using real shuffle statistics; on by default since Spark 3.2.
• Skew-join splitting — AQE breaking an oversized shuffle partition into sub-partitions so no single task holds it; the automatic form of key salting.
• Coalescing shuffle partitions — AQE merging tiny post-shuffle partitions into right-sized ones (~64MB target) to steady per-task memory and cut scheduler overhead.
• Dynamic join switching — AQE promoting a SortMergeJoin to a BroadcastHashJoin at runtime when a side proves small enough, removing shuffle-sort buffers.
• AdaptiveSparkPlan isFinalPlan=true — the plan wrapper indicating the runtime, AQE-revised plan that actually executed.

If this saved you a 3 AM pager incident, follow me on Medium for Twice a week for the next 3 Months of Advanced Data Engineering & AI. Let’s connect on LinkedIn and X.

Join thousands of data leaders on the AI newsletter. Join over 80,000 subscribers and keep up to date with the latest developments in AI. From research to projects and ideas. If you are building an AI startup, an AI-related product, or a service, we invite you to consider becoming a sponsor.

Published via Towards AI