Spark memory management: every region, and the arithmetic that sizes it

• Architecture: what the cluster manager allocates
• The four regions inside the heap
• Checking the arithmetic against a running executor
• The elastic boundary
• What I could and could not demonstrate
• Off-heap memory
• How tasks share execution memory
• The one hard floor: 450 MiB
• Tuning, in the order worth trying
• Production notes
• Frequently asked questions
• Conclusion
• References
• Trademarks

TL;DR

• An executor’s heap is split four ways: a fixed 300 MB reserved, a user region Spark does not manage, and one unified pool shared by execution and storage.
• The sizes come from two fractions applied in order: (heap - 300MB) * spark.memory.fraction, then * spark.memory.storageFraction for storage’s protected share. On a 4 GB heap that is 2277.6 MiB unified, of which 1138.8 MiB is storage’s to keep.
• Execution wins. The boundary between execution and storage is elastic and only storage gets evicted, so cached data is never guaranteed to still be there.
• Off-heap memory is additional, not carved out of the heap, and the container needs a third allocation on top of both: memory overhead, which is where JVM internals and Python live.
• The container is killed for the total, not for the heap. heap + off-heap + overhead + PySpark is the number your cluster manager enforces.

Memory is where Spark tuning stops being folklore and becomes arithmetic. Given three configuration values you can compute every region size exactly, and given the region sizes most memory failures become predictable rather than mysterious.

The trouble is that the regions have confusing names, two of them overlap, one is invisible to Spark, and the number the cluster manager kills you for is none of the ones you configured. This post lays out each region, derives its size, checks the derivation against a running executor, and then goes through what happens when each one fills up.

Every number below was read from a live session or from the Spark source. Where I tried to demonstrate something and failed, the post says so.

Architecture: what the cluster manager allocates

Start outside the JVM, because this is the level at which your job gets killed. A container is three or four allocations, only one of which is the heap:

flowchart TB
  subgraph C["executor container, what YARN or Kubernetes reserves"]
    H["<b>JVM heap</b><br/>spark.executor.memory"]
    O["<b>off-heap</b><br/>spark.memory.offHeap.size<br/>(0 unless enabled)"]
    V["<b>memory overhead</b><br/>spark.executor.memoryOverhead<br/>JVM internals, native buffers"]
    P["<b>PySpark memory</b><br/>spark.executor.pyspark.memory<br/>(only for Python)"]
  end
  C --> K{"total > container limit?"}
  K -->|"yes"| X["container killed<br/>by the cluster manager"]
  K -->|"no"| R["runs"]

The point of that diagram is the arrow at the bottom. spark.executor.memory sizes the heap, and nothing else. The cluster manager enforces the sum, so a job that raises executor.memory to exactly the container limit is asking to be killed, because overhead still has to fit somewhere.

So overhead is max(executor.memory * 0.10, 384m) unless you set it directly, in which case the factor and the minimum are both ignored. That is worth knowing because the two derived knobs look like they still apply and they do not.

A naming change worth pinning down. spark.executor.minMemoryOverhead carries .version("4.0.0") in the config definitions, and it is absent from the 3.5.9 source, where the same 384 MiB floor was the hardcoded constant ResourceProfile.MEMORY_OVERHEAD_MIN_MIB = 384L. The number has not changed; it became configurable.

The four regions inside the heap

Now inside the JVM. The heap is divided in a fixed order, and each division is a simple multiplication:

flowchart TB
  A["<b>JVM heap</b><br/>for example 4096 MiB"] --> B["<b>Reserved: 300 MB</b><br/>fixed, not configurable"]
  A --> C["<b>usable = heap - 300MB</b><br/>3796 MiB"]
  C --> D["<b>User memory</b><br/>usable x (1 - memory.fraction)<br/>1518 MiB, unmanaged"]
  C --> E["<b>Unified pool</b><br/>usable x memory.fraction<br/>2277.6 MiB"]
  E --> F["<b>Storage</b><br/>protected share:<br/>pool x storageFraction<br/>1138.8 MiB"]
  E --> G["<b>Execution</b><br/>the rest, and it can<br/>take storage's free space"]

Each region has a different job and a different failure mode:

The user region is the one that surprises people. Spark does not track it, so it has no spill path and no eviction. If your UDF builds a large dictionary per task, it competes for the 40% of usable heap that Spark is deliberately not managing, and the failure is a plain OutOfMemoryError.

Checking the arithmetic against a running executor

The derivation above is easy to get wrong, so here it is checked against the MemoryManager in a live session with a 4 GB heap:

val mm = org.apache.spark.SparkEnv.get.memoryManager
def mb(b: Long) = f"${b / 1024.0 / 1024.0}%.1f MiB"

println("runtime maxMemory     = " + mb(Runtime.getRuntime.maxMemory))
println("maxOnHeapStorage      = " + mb(mm.maxOnHeapStorageMemory))
println("memory manager class  = " + mm.getClass.getSimpleName)

// the documented arithmetic, recomputed by hand
val heap     = Runtime.getRuntime.maxMemory
val usable   = heap - 300L * 1024 * 1024
val unified  = (usable * 0.6).toLong
println("usable  (heap - 300MB) = " + mb(usable))
println("unified (x 0.6)        = " + mb(unified))
println("storage region (x 0.5) = " + mb((unified * 0.5).toLong))

runtime maxMemory     = 4096.0 MiB
maxOnHeapStorage      = 2277.6 MiB
memory manager class  = UnifiedMemoryManager

usable  (heap - 300MB) = 3796.0 MiB
unified (x 0.6)        = 2277.6 MiB
storage region (x 0.5) = 1138.8 MiB

The hand calculation matches maxOnHeapStorageMemory exactly. And that raises the point most diagrams get wrong.

maxOnHeapStorageMemory is 2277.6 MiB, the whole unified pool, not the 1138.8 MiB storage region. Storage is not capped at its fraction. The fraction sets how much storage may keep under pressure; with execution idle, cached data can occupy the entire pool. The two numbers answer different questions:

• spark.memory.storageFraction is the share of the pool that execution cannot take back.
• maxOnHeapStorageMemory is the share that storage may grow into when nothing else wants it.

Read storageFraction as an eviction floor, not a size limit. That single reframing makes the next section obvious.

The elastic boundary

Execution and storage share one pool, and the divide between them moves. The rules are asymmetric and they are short enough to read directly.

flowchart TB
  S1["storage using less<br/>than its share"] -->|"execution needs memory"| S2["execution takes<br/>the free space"]
  S3["storage borrowed past<br/>its protected share"] -->|"execution needs memory"| S4["blocks are <b>evicted</b><br/>until storage is back<br/>to its share"]
  E1["execution using more<br/>than half the pool"] -->|"storage needs memory"| E2["storage <b>waits</b>.<br/>Execution is never evicted,<br/>it only spills on its own terms"]

From UnifiedMemoryManager, this is what execution is allowed to reclaim:

        val memoryReclaimableFromStorage = math.max(
          storagePool.memoryFree,
          storagePool.poolSize - storageRegionSize)

The max of two things: storage’s free space, and however much storage has borrowed beyond its region. So execution can always take back what it lent, plus anything storage is not currently using.

And this is the cap on how large the execution pool may become:

    def computeMaxExecutionPoolSize(): Long = {
      maxMemory - math.min(storagePool.memoryUsed, storageRegionSize)
    }

min(storage used, storage region) is exactly the protected part. Execution may grow into everything except that.

There is no symmetric rule for storage. Nothing in the manager evicts execution memory, because a half-built hash table cannot be dropped and recomputed cheaply the way a cached block can. Execution releases memory by spilling to disk, and it decides when.

The practical consequence: cache() is a hint, not a guarantee. A cached DataFrame that fitted yesterday can be partly evicted today because a different stage in the same application wanted execution memory.

What I could and could not demonstrate

Two claims from the section above are easy to state and harder to produce on demand, so here is what actually happened when I tried.

Storage borrowing past its region: confirmed. With a 1 GB driver heap, so a 434 MiB unified pool and a 217 MiB storage region, caching 7,000,000 rows put 306 MiB into storage across 8 blocks, comfortably past the region:

pool=434 MiB  region=217 MiB
after caching: storage=306 MiB blocks=8  beyond region=true

Eviction under execution pressure: not reproduced. I then ran sorts designed to demand execution memory, up to 30,000,000 rows over 8 concurrent tasks, and nothing was evicted:

after sort  : storage=306 MiB blocks=8  evicted=0
peakExecutionMemory (max task) = 0 MiB
memoryBytesSpilled  (total)    = 0 MiB
diskBytesSpilled    (total)    = 0 MiB

Zero spill is the explanation: there was never real pressure, so nothing needed reclaiming. The mechanism is in the source above and I am not going to claim I watched it fire when I did not. If you want to see eviction on your own cluster, the observable signals are the storage memory dropping in the executors tab and cached partitions falling below 100% in the storage tab.

A different way cache fails, and this one did reproduce. Caching more than the pool can hold with MEMORY_ONLY does not raise anything. It silently caches nothing:

# 20,000,000 rows into a 434 MiB pool
after caching: storage=0 MiB  blocks=0
cached count = 20000000     # still correct, recomputed from source

Zero blocks cached, and the count is still right because Spark recomputed the whole thing. A cache() call that achieves nothing while your job still produces correct answers is the worst kind of performance bug, because nothing fails. The storage tab showing a cached RDD at 0% is the tell, and MEMORY_AND_DISK is the usual fix.

Off-heap memory

Off-heap is Tungsten allocating with sun.misc.Unsafe outside the Java heap, so the data is not subject to garbage collection and is stored in Spark’s own binary format rather than as Java objects.

// started with --conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=1g
println("maxOnHeapStorage   = " + mb(mm.maxOnHeapStorageMemory))
println("maxOffHeapStorage  = " + mb(mm.maxOffHeapStorageMemory))
println("tungstenMemoryMode = " + mm.tungstenMemoryMode)

maxOnHeapStorage   = 2277.6 MiB
maxOffHeapStorage  = 1024.0 MiB
tungstenMemoryMode = OFF_HEAP

Three things to take from that output:

• Off-heap is additional. The on-heap pool is unchanged at 2277.6 MiB. Enabling 1 GB off-heap did not shrink the heap; it added a second pool, and the container now needs 1 GB more than before.
• It has its own execution and storage split, governed by the same storageFraction, so the same elastic rules apply within it.
• tungstenMemoryMode flips to OFF_HEAP, which is what actually changes where sorts and hash tables allocate.

The trade is GC pressure against manual accounting. Off-heap memory is invisible to the JVM, so a leak or an underestimate shows up as the cluster manager killing the container rather than as an OutOfMemoryError you can read a stack trace from.

How tasks share execution memory

One more division, inside the execution pool. Several tasks run per executor and they compete, with a fairness rule in ExecutionMemoryPool:

      val maxPoolSize = computeMaxPoolSize()
      val maxMemoryPerTask = maxPoolSize / numActiveTasks
      val minMemoryPerTask = poolSize / (2 * numActiveTasks)

flowchart TB
  P["execution pool, N active tasks"] --> A["each task capped at <b>1/N</b><br/>of the pool"]
  P --> B["each task guaranteed at least <b>1/2N</b><br/>before it is made to wait"]
  B --> C["below 1/2N, the task blocks<br/>until another releases memory"]

So a task can never take more than 1/N of the pool, and it will not be blocked until it has at least 1/2N. N is the number of active tasks, not the core count, so the guarantee moves as tasks start and finish.

This is why spark.executor.cores is a memory setting as much as a parallelism setting. Doubling the cores per executor halves each task’s share of the same pool, which is the usual reason a job that worked with 4 cores per executor spills constantly with 8.

The one hard floor: 450 MiB

Spark refuses to start if the heap cannot cover the reserved region with room to spare. From UnifiedMemoryManager:

  private val RESERVED_SYSTEM_MEMORY_BYTES = 300 * 1024 * 1024
  ...
    val minSystemMemory = (reservedMemory * 1.5).ceil.toLong

300 MB * 1.5 is 450 MiB, and there are two separate checks. A 400 MB driver:

[INVALID_DRIVER_MEMORY] System memory 419430400 must be at least 471859200.

and a 400 MB executor:

[INVALID_EXECUTOR_MEMORY] Executor memory 419430400 must be at least 471859200.

471859200 is exactly 450 MiB. This is a fail-fast check added deliberately, and it is a good thing: below that size the unified pool would be a few tens of megabytes and every job would spill pathologically instead of failing clearly.

Tuning, in the order worth trying

The ordering matters more than any individual row. Overhead problems masquerade as heap problems, and the first question for any kill is always whether the JVM died or the container was terminated from outside.

Production notes

• Size the container, not the heap. heap + off-heap + overhead + PySpark is the number enforced, so leave headroom rather than setting executor.memory to the limit.
• Setting spark.executor.memoryOverhead directly disables both derived knobs. The factor and the minimum are ignored once it is explicit.
• Treat storageFraction as an eviction floor. It does not cap how much cache can grow, only how much survives pressure.
• Never assume a cached DataFrame is resident. Check the storage tab for the cached fraction before attributing a slow stage to something else.
• Use MEMORY_AND_DISK unless you have measured that the data fits. MEMORY_ONLY silently caches nothing when it does not, and the job stays correct while getting slower.
• Treat spark.executor.cores as a memory knob. It divides the execution pool 1/N, so raising it can cause spill with no other change.
• Account for off-heap twice. Once in offHeap.size and once in the container total. It is additional to the heap, not carved from it.
• Do not tune spark.memory.fraction first. It trades user memory against managed memory, and the common problems are overhead and core count.

Frequently asked questions

Why is my container killed when the heap looks fine? Because the cluster manager enforces the total, and overhead is not part of spark.executor.memory. A heap sitting at 60% with off-heap and Python pushing the container over its limit gets killed with no JVM OutOfMemoryError at all, usually as exit code 137.

What is actually in the 300 MB reserved region? Nothing of yours. It is a margin so that Spark’s own internal structures cannot be squeezed out by execution and storage. It is not configurable outside tests, and the 450 MiB startup floor exists to keep it meaningful.

Does spark.memory.storageFraction limit how much I can cache? No, and this is the most common misreading. It sets the portion of the unified pool that execution cannot evict. With execution idle, cache can occupy the whole pool, which is why maxOnHeapStorageMemory reports the full pool size.

Can storage evict execution? No. The asymmetry is deliberate: a cached block can be dropped and recomputed, while a partially built hash table cannot. Execution gives memory back by spilling, on its own schedule.

Is off-heap memory faster? It is not automatically faster. It removes garbage-collection pressure and stores data compactly, which helps on large heaps, and it costs you the JVM’s memory accounting. A leak becomes a killed container instead of a stack trace.

Where does PySpark memory come from? Outside the JVM, from the overhead allocation, unless you set spark.executor.pyspark.memory to reserve it explicitly. A Python UDF holding a large object is not visible in any Spark memory metric.

Why did my job start spilling after I gave executors more cores? Because each task is capped at 1/N of the execution pool where N is the active task count. More cores means more concurrent tasks and a smaller share each, from the same pool.

Conclusion

The useful way to hold all of this is as one sum and one asymmetry.

The sum is the container: heap, off-heap, overhead and Python, of which spark.executor.memory is only the first. Most memory incidents that look mysterious are the cluster manager enforcing that sum while you were watching the heap.

The asymmetry is inside the unified pool: execution can take memory from storage, and storage can never take it from execution. Every consequence follows from that. Cached data is evictable, so caching is a hint. storageFraction is a floor rather than a ceiling, so it does not limit cache growth. Execution handles its own shortfall by spilling, which is why spill is a tuning signal rather than an error. Once those two ideas are in place the configuration stops being a list of fractions to try and becomes arithmetic you can do before changing anything: compute the pool from the heap, divide by the concurrent task count, and compare it with what one task actually needs. That prediction beats trial and error.

References

• Spark configuration reference for the memory and overhead settings and their defaults
• UnifiedMemoryManager.scala, which holds the 300 MB constant, the 450 MiB check and the borrowing rules quoted above
• ExecutionMemoryPool.scala for the per-task 1/N cap and the 1/2N guarantee
• Dive into Spark memory, whose region-by-region breakdown is the shape this post follows
• Apache Spark architecture for the executor and task machinery these pools belong to

Trademarks

Apache Spark, Apache Hadoop and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.