Skip to main content
Apache Spark advanced Lesson 6 of 6

Spark Memory Management and Debugging Failures

Read Spark's memory model, then reproduce and fix the three failures that actually take jobs down — driver OOM, executor OOM, and disk spill.

Spark failures are mostly memory failures, and they are legible once you know which pool ran out. This lesson reproduces each one.

The memory model

An executor’s JVM heap is divided:

┌─────────────────────────────────────────────────────────┐
│ Executor JVM heap (spark.executor.memory = 8g)          │
│                                                         │
│  ┌───────────────────────────────────────────────────┐  │
│  │ Reserved (300 MB, fixed)                          │  │
│  └───────────────────────────────────────────────────┘  │
│  ┌───────────────────────────────────────────────────┐  │
│  │ Unified pool  (spark.memory.fraction = 0.6)       │  │
│  │  ┌─────────────────────┬──────────────────────┐   │  │
│  │  │ Storage             │ Execution            │   │  │
│  │  │ cached blocks       │ shuffles, joins,     │   │  │
│  │  │ broadcast vars      │ sorts, aggregations  │   │  │
│  │  │ (evictable)         │ (not evictable)      │   │  │
│  │  └─────────────────────┴──────────────────────┘   │  │
│  │        boundary moves — they borrow from each other│  │
│  └───────────────────────────────────────────────────┘  │
│  ┌───────────────────────────────────────────────────┐  │
│  │ User memory (0.4) — your objects, UDF state       │  │
│  └───────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘
        + off-heap overhead (memoryOverhead), outside the heap

The asymmetry is the part that matters: execution can evict storage, storage cannot evict execution. A large shuffle will silently drop your cached DataFrame; a large cache will never starve a shuffle.

Check the actual numbers:

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder.appName("memory").master("local[2]")
         .config("spark.driver.memory", "2g")
         .config("spark.executor.memory", "2g")
         .getOrCreate())

for k in ["spark.memory.fraction", "spark.memory.storageFraction",
          "spark.executor.memory", "spark.driver.memory"]:
    print(f"{k:32} {spark.conf.get(k, '(default)')}")
spark.memory.fraction            0.6
spark.memory.storageFraction     0.5
spark.executor.memory            2g
spark.driver.memory              2g

Failure 1: driver OOM from collect

big = spark.range(50_000_000).select(
    F.col("id"), (F.rand() * 1000).alias("v"), F.concat(F.lit("row-"), F.col("id")).alias("label")
)

rows = big.collect()
25/02/15 12:04:18 ERROR TaskSetManager: Total size of serialized results of 8 tasks (2.1 GB)
is bigger than spark.driver.maxResultSize (1024.0 MB)

py4j.protocol.Py4JJavaError: An error occurred while calling o56.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure:
Total size of serialized results is bigger than spark.driver.maxResultSize

Spark caught this one before the heap actually blew, because maxResultSize defaults to 1 GB. Raising it just moves the failure:

spark.conf.set("spark.driver.maxResultSize", "8g")
rows = big.collect()
java.lang.OutOfMemoryError: Java heap space
	at java.base/java.util.Arrays.copyOf(Arrays.java:3537)
	at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3715)

The fix is not more driver memory — it is not collecting:

print("count:", big.count())                       # aggregate on executors
big.limit(5).show()                                # bounded
big.write.mode("overwrite").parquet("/tmp/big")    # write from executors
count: 50000000
+---+------------------+--------+
| id|                 v|   label|
+---+------------------+--------+
|  0| 412.331276160446| row-0  |
|  1| 887.204718641342| row-1  |
|  2|  91.556632900231| row-2  |
|  3| 233.901128845629| row-3  |
|  4| 764.331921830123| row-4  |
+---+------------------+--------+

toPandas() has the same problem and is worse, because it materialises the data twice — once as JVM rows, once as Python objects. Use spark.sql.execution.arrow.pyspark.enabled to halve that, and still only on small results.

Failure 2: executor OOM from a wide partition

skewed = spark.range(20_000_000).select(
    F.lit(1).alias("key"),                      # every row shares one key
    F.concat(F.lit("payload-"), F.col("id")).alias("payload"),
)

skewed.groupBy("key").agg(F.collect_list("payload")).count()
25/02/15 12:19:33 ERROR Executor: Exception in task 0.0 in stage 4.0 (TID 62)
java.lang.OutOfMemoryError: Java heap space
	at org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder.grow
	at org.apache.spark.sql.catalyst.util.GenericArrayData

collect_list builds one array holding 20 million strings in a single executor’s heap. No partition count or memory setting fixes this, because the result does not fit. Aggregate to something bounded instead:

skewed.groupBy("key").agg(
    F.count("*").alias("n"),
    F.first("payload").alias("sample"),
).show()
+---+--------+---------+
|key|       n|   sample|
+---+--------+---------+
|  1|20000000|payload-0|
+---+--------+---------+

The rule: any aggregate whose output grows with input size — collect_list, collect_set, a pivot with unbounded cardinality — will eventually fail on a large group.

Failure 3: spill

Spill is survivable but slow. Force it by shrinking partition count:

import time

data = spark.range(30_000_000).select(
    (F.col("id") % 500_000).alias("key"), F.rand().alias("v")
)

for parts in (4, 64, 400):
    spark.conf.set("spark.sql.shuffle.partitions", parts)
    t0 = time.perf_counter()
    data.groupBy("key").agg(F.avg("v")).count()
    print(f"{parts:4} partitions: {time.perf_counter() - t0:6.2f}s")
   4 partitions:  41.83s
  64 partitions:  11.27s
 400 partitions:   8.94s

With four partitions each task held 7.5 million rows, exceeded its execution memory, and wrote sorted runs to disk. The Spark UI shows it directly in the stage detail:

Metric                    Min      Median    Max
Duration                  38 s     40 s      42 s
GC Time                   4 s      5 s       6 s
Shuffle Read Size         180 MB   182 MB    184 MB
Spill (Memory)            2.1 GB   2.2 GB    2.3 GB
Spill (Disk)              610 MB   640 MB    668 MB

Non-zero Spill (Disk) is the signal. Fix it with more partitions before reaching for more memory — partitions are free, memory is not.

Cache eviction

cached = spark.range(20_000_000).select(F.col("id"), F.rand().alias("v")).cache()
cached.count()
print("cached partitions:", spark.sparkContext._jsc.sc().getRDDStorageInfo().length)
cached partitions: 1

Now run a shuffle-heavy job in the same session and check again:

data.groupBy("key").agg(F.avg("v")).count()          # heavy shuffle
print(cached.storageLevel)
print("still cached fully?", cached.count())
Disk Memory Deserialized 1x Replicated
still cached fully? 20000000

The count is still right, but some blocks were evicted from memory and re-read from disk — the MEMORY_AND_DISK level means correctness is preserved and only speed suffers. Watch the Storage tab’s “Fraction Cached” to see this happening; a cache that drops below 100% during heavy stages is execution memory reclaiming its space.

To protect a cache, raise spark.memory.storageFraction — but that shrinks what execution can borrow, so it trades one failure mode for another.

Reading GC pressure

spark-submit \
  --conf "spark.executor.extraJavaOptions=-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps" \
  app.py ...
[GC (Allocation Failure) [PSYoungGen: 1398144K->174520K(1567744K)] 3947264K->3512392K(5138432K), 0.4821 s]
[Full GC (Ergonomics) [PSYoungGen: 174520K->0K(1567744K)] [ParOldGen: 3337872K->2891244K(3570688K)] 3512392K->2891244K(5138432K), 3.8214 s]

A 3.8-second full GC pause, repeated, means the heap is too full. In the UI, compare GC Time to Duration per task — above roughly 10% is a problem. The fixes, in order of preference:

  1. More partitions, so each task’s working set is smaller
  2. Less caching, or MEMORY_AND_DISK_SER to store compactly
  3. Fewer cores per executor, giving each task more heap
  4. Only then, more executor memory

A diagnostic table

SymptomWhereCauseFix
maxResultSize exceededdrivercollect() on large dataaggregate or write instead
OutOfMemoryError in collectFromPlandriversame, without the guardsame
OutOfMemoryError in GenericArrayDataexecutorunbounded collect_listbounded aggregate
Container killed by YARNexecutoroff-heap overflowraise memoryOverhead
Non-zero Spill (Disk)executorpartitions too largemore partitions
GC time > 10% of durationexecutorheap pressuremore partitions, less cache
One task far slower than medianexecutorskewsalt the key, or broadcast

Practice

1. Call collect() on a 10-million-row DataFrame with a 1 GB driver. 
Total size of serialized results of 8 tasks (1740.2 MB) is bigger than
spark.driver.maxResultSize (1024.0 MB)

Spark’s guard fired before the heap did. Treat maxResultSize as a safety rail, not a limit to raise — the fact that you hit it means the design is wrong.

2. Run a groupBy with 4 and then 400 shuffle partitions, comparing Spill (Disk). 
  4 partitions: Spill (Disk) 640 MB, duration 41.8s
400 partitions: Spill (Disk)   0 B,  duration  8.9s

Eliminating spill accounted for nearly all the speed-up. Partition count is almost always the first thing to change when a shuffle stage is slow.

3. Cache a DataFrame, then run a large shuffle. Check Fraction Cached. 
before shuffle: 100% cached, 1.4 GB in memory
after shuffle:   62% cached, 0.9 GB in memory, 0.5 GB on disk

Execution memory evicted storage blocks. Nothing failed and no warning was printed — the cache just got slower. This is the clearest illustration of the one-way eviction rule.

4. Set spark.memory.fraction to 0.2 and run a shuffle-heavy job. 
default (0.6): 8.94s,  Spill (Disk) 0 B
lowered (0.2): 36.71s, Spill (Disk) 1.9 GB

Shrinking the unified pool starves execution and forces spill, while the 0.4 user-memory portion sits mostly unused. The default is well chosen — lower it only if your UDFs genuinely allocate large amounts of user memory.

That completes the Spark core track: architecture, RDDs, the SQL engine, deployment, and the memory model behind every failure you will meet.

Frequently Asked Questions

What is unified memory management?
Execution memory (shuffles, joins, sorts) and storage memory (cached data) share one pool and borrow from each other. Execution can evict cached blocks when it needs room; storage cannot evict execution. That is why a cache silently shrinks during a big shuffle.
Why does my driver run out of memory when executors are fine?
Something brought data to the driver — collect(), toPandas(), a large broadcast, or too many task results accumulating. The driver holds the plan and the results you request, not the distributed dataset.
Is disk spill always bad?
No. Spill is Spark degrading gracefully instead of failing, and a job that spills a little may still be fine. Heavy repeated spill means partitions are too large — increase partition count before increasing memory, since it is usually cheaper.
How do I read a garbage collection problem?
Compare GC time to task duration in the Spark UI's stage detail. GC above about 10% of task time means the heap is under pressure. The usual fixes are more partitions, less caching, or fewer cores per executor so each task has more headroom.