Spark Executor Memory

A task never gets the executor's memory. It gets the executor's memory divided by the number of tasks running on that executor, after two fixed deductions. Every confusing thing about Spark memory — why more cores made the job spill, why raising memory did nothing, why the container was killed while the heap looked fine — comes out of that one division and the two deductions in front of it.

The layout is fixed and comes from UnifiedMemoryManager. Start with the Java heap, which is what spark.executor.memory sets. Subtract a hard-coded 300 MiB reserve. Of what is left, spark.memory.fraction (default 0.6) is the unified region shared by execution and storage; the other 40% is user memory, for the objects your own code and Spark's internal metadata allocate. Inside the unified region, spark.memory.storageFraction (default 0.5) marks how much cached data is immune from eviction.

The panel computes that layout and then divides it among your tasks. The control to move first is spark.executor.cores, because it changes the answer without changing a single byte of memory.

Leave everything at its default and drag spark.executor.cores from 4 to 16. The heap has not changed by one byte, but the ceiling per task falls from 1.2 GiB to 296 MiB, and tasks that fitted comfortably now spill. This is why "give the executor more cores, it has plenty of memory" is a change to the memory configuration, whether or not anyone meant it that way.

The arithmetic, once, with numbers

For an 8 GiB executor at the defaults:

heap                 8192 MiB   (spark.executor.memory)
- reserve             300 MiB   (a constant in the source, not configurable)
= usable             7892 MiB
x 0.60               4735 MiB   unified region (execution + storage)
x 0.40               3157 MiB   user memory (your objects, Spark's own structures)

storage region       2368 MiB   = 4735 x storageFraction 0.5, immune from eviction
per-task ceiling     1184 MiB   = execution pool 4735 / 4 concurrent tasks
per-task guarantee    592 MiB   = pool / (2 x 4), the point below which a task blocks

Two of those lines are worth arguing with. The 300 MiB reserve is RESERVED_SYSTEM_MEMORY_BYTES, a literal constant; an executor configured with less than 1.5× it — 450 MiB — refuses to start at all, with an INVALID_EXECUTOR_MEMORY error. And user memory is not spare — it holds anything your user-defined functions allocate, the objects a mapPartitions builds, and Spark's own bookkeeping. Raising spark.memory.fraction to 0.8 takes that space, and the symptom of taking too much is a heap OutOfMemoryError in code that has nothing to do with shuffling.

Execution and storage share, but not symmetrically

The unified region has a soft boundary, and the asymmetry is the part people get wrong. Both directions of borrowing are allowed, but only one direction of taking-back is.

• Storage borrows from execution, and when execution wants that memory back, cached blocks are evicted until it has enough. Your cached DataFrame quietly loses partitions and gets recomputed.
• Execution borrows from storage, and storage can never take it back. The source comment gives the reason plainly: implementing it is complex. So a cache() that arrives while a shuffle is running may simply fail to store, and those blocks go straight to their fallback storage level.

spark.memory.storageFraction is the floor under the first bullet: storage below that line is immune. The exact expression is maxMemory − min(storageUsed, storageRegionSize), and it is the execution pool's size. Set cached data to 3 GiB in the panel with the defaults: 2.3 GiB of it is protected, so the execution pool shrinks from 4.6 GiB to 2.3 GiB and every task's ceiling halves. Caching is not free even when it looks like it fits.

Now drag storageFraction to 0. Nothing is immune, the cache is fully evictable, and execution gets the whole region back. That is the correct setting for a job that caches nothing, and the reason the tuning guide still tells you to leave it at 0.5 is that a job which caches something and then recomputes it repeatedly is usually slower than one that spills.

Spilling is the design, not the failure

When a task reaches its ceiling, it does not fail. The sorter it is using — UnsafeExternalSorter for a sort, ShuffleExternalSorter for a shuffle write — sorts what it holds, writes that run to local disk, and frees the memory. At the end it merges the runs. A task with a 1.2 GiB ceiling and a 6 GiB working set does this five times.

So spill is a slowdown, not an error, and the amount is bounded by disk rather than by memory. Two things do turn it into an error. The first is a single record, or a single group of records held together, that cannot fit in the ceiling at all — a 2 GiB array built by a collect_list over one hot key has nowhere to go. The second is user memory, which has no spilling mechanism: a Java object you allocated is either in the heap or the JVM throws.

This is exactly the distinction the Spark UI blurs. A stage with heavy spill and a stage that OOMs look like the same "not enough memory" problem, and only one of them is.

Why more memory hides skew instead of fixing it

Here is the whole argument in one line: raising spark.executor.memory raises the ceiling for every task by the same factor, and skew is about one task having more work, which the ceiling does not touch.

The panel below runs the stage. Four tasks per executor, one of them holding the hot partition. Drag the memory slider and watch two numbers move in opposite ways.

At 8 GiB and 20× skew the stage takes 2.2 minutes and the hot task spills 6.7 GiB. Drag to 32 GiB and it is 1.4 minutes with 3.1 GiB of spill; at 64 GiB the spill readout says none, which in the Spark UI reads as a solved problem. The wall-clock is then 40 seconds and stops improving no matter how much more memory you add, because one core still has to read 7.8 GiB of rows. That floor is the fourth readout, and eight times the memory bought you 2.2 minutes down to 40 seconds — while 69% of the stage's core-time is still idle.

Now tick split the hot partition into 8. Wall-clock falls by roughly the factor you split by, at 8 GiB, with the spill still happening. Splitting the work beats enlarging the container, every time, and the two are not substitutes. That splitting is what salting does by hand and what adaptive query execution does automatically — and it is exactly what neither can do when the partition is a single key, which is the wall described in the partitioning lesson.

The memory that is not in the heap

A cluster manager does not allocate you a heap. It allocates a container, and the container is:

container = spark.executor.memory + spark.executor.memoryOverhead
memoryOverhead defaults to max(0.10 x spark.executor.memory, 384 MiB)

The overhead holds everything the JVM allocates outside its heap: thread stacks, the metaspace, Netty's off-heap network buffers for shuffle transfers, memory-mapped files, and — the big one — Python worker processes for PySpark. On Kubernetes the factor defaults to 0.40 rather than 0.10 for non-JVM jobs, precisely because Python breaks the 10% assumption.

This is a different budget with a different failure. Exceed the heap and the JVM throws java.lang.OutOfMemoryError with a stack trace. Exceed the container and the cluster manager kills the process from outside: Container killed by YARN for exceeding memory limits. 9.2 GB of 9.0 GB physical memory used, or on Kubernetes an ExecutorLostFailure with exit code 137, which is the kernel OOM killer rather than the JVM. Raising spark.executor.memory in response to the second message makes the container bigger and the heap bigger, so it sometimes appears to work — but if the pressure is Python or network buffers, you have moved the boundary rather than the usage, and the same message returns at a larger scale.

Diagnosing it in your own cluster

Read these four in order, and stop at the first one that answers you.

• Spill columns, in the stage's task summary. The Spark UI stage page has Spill (Memory) and Spill (Disk) rows with Min / 25th / Median / 75th / Max. If Median is 0 and Max is large, one task is spilling and you have skew, not a memory shortage. If every quartile is large, the whole stage is too big for its budget and memory genuinely helps.
• The ratio in the same table. Compare Max to Median for Shuffle Read Size / Records. Skew is a ratio, and it is the only number that tells you whether splitting or enlarging is the right fix.
• The executor page. Look at Peak Execution Memory against your computed ceiling, and at GC Time as a share of task time. GC above about 10% of task time is a user-memory problem, not an execution one — the fix is fewer or smaller objects, or a lower spark.memory.fraction, not a bigger heap.
• The kill message itself. If it names physical memory and a container size, the heap is irrelevant; raise spark.executor.memoryOverhead. If it is a Java OutOfMemoryError, note whether the stack is inside a sorter (execution, spillable, so the real cause is one indivisible record or group) or inside your own code (user memory, not spillable at all).

Next: where the oversized partition came from, in the shuffle; how rows were assigned to it, in partitioning; and the other budget a bad size estimate can blow up, in the broadcast join and driver memory.