Spark Heap OutOfMemory Fix: Driver vs Executor

Spark OutOfMemoryError: Java heap space splits into two diseases sharing one name. Driver OOM kills the coordinator — the single JVM running your main() — usually because collect(), toPandas(), or an oversized broadcast dragged cluster-scale data onto one machine. Executor OOM kills workers — the JVMs running tasks — usually because one partition holds 40% of the data (skew), partitions number in the dozens for terabytes, or a shuffle materializes giants. Raising the wrong memory pool is the classic expensive no-op.

The economics punish guessing. Doubling executor memory across 50 nodes costs 50x the RAM while the skewed task still holds its hot key; doubling driver memory for a skewed join fixes nothing while the bill climbs. Classification is cheap — the Spark UI's stage detail names the victim (driver stderr versus failed tasks) and the task histogram names the shape (one giant versus many larges) — and it directs money exactly where it helps.

This guide builds the split-brain playbook: driver discipline (bounded takes, Parquet sinks, broadcast budgets), executor discipline (salting, repartitioning, file sizing), memory math per pool, and GC-log reading that separates churn from genuine growth. Adaptive execution gets honest credit for what it automates and clear limits where it stops.

Driver vs Executor: Read the Victim First

Spark runs two JVM roles with separate heaps and separate failure modes. The driver plans, coordinates, and collects — one JVM, modest heap (4-8G), killed by anything that centralizes data: collect, toPandas, takeOrdered on millions, giant broadcasts, or driver-side Python UDF accumulation. The executors execute — many JVMs, larger heaps, killed by anything that concentrates data on one of them: skewed keys, fat partitions, exploding joins, or cached DataFrames never unpersisted.

The UI names the victim in seconds. Driver OOM: the application dies with driver stderr naming heap space while completed stages show green executors — nothing failed distributedly because the coordinator fell over. Executor OOM: stages show red failed tasks with heap space in task stderr, often retrying 4 times (spark.task.maxFailures) before the job aborts. Same exception text, opposite funerals — read the stage colors before the logs.

This split dictates every subsequent action. Driver victims get result-size discipline and broadcast budgets; executor victims get skew analysis and partition math. Cross-applying wastes both time and money: 16G executors can't digest an 800GB hot key, and driver raises can't drain a skewed shuffle. The first minute of every heap incident belongs to victim identification, and the UI answers it faster than any log.

# Victim ID from the outside (YARN example)
# yarn logs -applicationId application_1710000000000_0042 | grep -B3 -A8 'java.lang.OutOfMemoryError' | head -n 40
# Driver corpse: driver stderr names collectResult / TorrentBroadcast
# Executor corpse: task stderr names shuffle read / deserialization
# In-app guard: never trust an unbounded action
# df.write.mode('overwrite').parquet('s3://out/x/')  # sink, not collect

Driver Discipline: Stop Feeding the Chef

The driver coordinates; it must never hold data at cluster scale. collect() materializes every row in driver memory — 10M rows that executors handled effortlessly become a single 8GB driver list that kills a 4G heap instantly. toPandas() doubles the insult with Arrow conversion overhead. takeOrdered(n) on huge n sorts distributedly then ships everything to one machine. Each has a bounded twin: limit(n).collect() for inspection, write.parquet for sinks, approxQuantile for statistics.

Broadcasts are the stealth centralization. Broadcast joins copy the small side to every executor through the driver — a 2GB dimension forced past the 10MB auto threshold crushes driver memory during TorrentBroadcast, then punishes executors holding the copy. Measure dimensions before broadcasting (count plus avg row bytes), respect spark.sql.autoBroadcastJoinThreshold, and hint MERGE for large-side joins explicitly. Driver-side Python accumulation (appending batches to a list across foreach) dies the same death in slower motion.

Enforce with bans, not advice. Block df.collect() and toPandas() without limit() in review linters, cap driver memory at honest values (raising it masks centralization instead of fixing it), and alert on driver heap usage crossing 80%. A driver that stays thin is a coordinator doing its job; a driver that grows is a data sink wearing a coordinator's badge.

from pyspark.sql import functions as F
# BAD: entire result onto one machine
# rows = df.collect()
# pdf = df.toPandas()
# GOOD: inspect bounded, sink distributed
sample = df.limit(1000).collect()
df.write.mode("overwrite").parquet("s3://out/events/")
stats = df.approxQuantile("amount", [0.5, 0.95, 0.99], 0.01)
# Broadcast only measured-small dimensions
small = spark.read.parquet("s3://dim/country/")
print(small.count(), len(small.columns))  # verify < threshold first

Executor Skew: Salt the Hot Key

Skew means one key owns a wildly disproportionate share — the default uid with 40% of events, the null key every malformed row shares, the celebrity node in a graph. Hash partitioning sends each key to exactly one partition, so the hot key's 800GB lands on one executor regardless of cluster size. Adding nodes or memory can't help: the atomic unit (one key) exceeds the memory unit (one heap). Only splitting the key across buckets restores parallelism.

Salting spreads the hot key over N sub-keys: append a random bucket (0-31) to the hot side and explode the dimension side across all buckets, join on (key, bucket), then drop the salt. The 800GB fans out to 32 tasks of 25GB each — digestible, parallel, and fast (41 minutes to 3). Salt only the hot keys when dimensions are large (targeted salting via a hot-key list) to avoid exploding the entire join; salt everything when simplicity beats the extra shuffle.

Detect before detonating. The top-key share query (groupBy-count-orderBy-limit) belongs in every pipeline's pre-join gate — alert over 10% single-key share. The task histogram confirms live: max duration 10-50x the median with matching shuffle-read disparity. AQE's skew hints mitigate some cases automatically, but atomic hot keys at 40% sail through — salt explicitly and keep AQE as the second net, never the plan.

from pyspark.sql import functions as F
N = 32
# Fact side: random bucket per row
salted = fact.withColumn("salt", (F.rand() * N).cast("int"))
# Dimension side: replicate each row across all buckets
exploded = dim.withColumn("salt", F.explode(F.array(*[F.lit(i) for i in range(N)])))
joined = salted.join(exploded, on=["uid", "salt"]).drop("salt")
# Targeted variant: salt ONLY the known hot key, plain-join the rest
# hot = fact.filter(F.col('uid') == HOT); cold = fact.filter(F.col('uid') != HOT)

Memory Math: Size Each Pool Deliberately

Driver and executor memory are independent budgets set for independent reasons. spark.driver.memory (4-8G typical) covers plans, broadcast assembly, and bounded results — size it to the largest legitimate collect plus broadcast headroom, then stop. spark.executor.memory (8-16G typical) covers task working sets — size it to measured peak partition size times 1.5, never to the hot key's fantasy size. spark.executor.memoryOverhead (10-25% extra) covers off-heap, shuffles, and Python workers — raise it for PySpark UDF-heavy stages where Python RSS, not JVM heap, is the killer.

Partition math precedes memory math. Target 128-256MB per partition: spark.sql.files.maxPartitionBytes for reads, repartition(N) or adaptive coalescing for shuffles, and spark.sql.shuffle.partitions (default 200, raise to 800-2000 for terabytes). Dozens of partitions over terabytes means gigabyte tasks no heap survives — fix the count before the bytes. Kryo serialization (spark.serializer=org.apache.spark.serializer.KryoSerializer) halves object graphs versus Java serialization for cached and shuffled data.

Cost discipline closes the loop. Record peak executor heap per stage from the UI's executor tab, size to peak-plus-half, and return unused memory (the 16G-to-8G rollback saved the incident's budget). Memory-optimized instances help genuine uniform largeness, never atomic skew — buy them for measured peaks, not for histograms with one giant. Every pool gets a number with a measurement behind it, reviewed quarterly as data grows.

# Deliberate pools: driver modest, executors measured, overhead for Python
# --conf spark.driver.memory=6g
# --conf spark.executor.memory=8g
# --conf spark.executor.memoryOverhead=2g
# --conf spark.sql.shuffle.partitions=800
# --conf spark.serializer=org.apache.spark.serializer.KryoSerializer
# --conf spark.sql.adaptive.enabled=true
# --conf spark.sql.adaptive.skewJoin.enabled=true
# Verify peaks per stage in Spark UI Executors tab before resizing

GC Logs: Churn vs Growth in Ten Minutes

Garbage-collector logs separate allocation churn (fast young GC, healthy old gen) from genuine growth (old gen climbing monotonically). Enable -XX:+PrintGCDetails with -Xloggc on one executor, run the failing stage, and plot: sawtooth with full recovery means churn — too many temporary objects per row, usually Python UDFs or deserialization loops. Staircase without recovery means retention — cached DataFrames, leaked broadcasts, or accumulating driver lists.

Churn fixes attack object rates: replace Python UDFs with Spark SQL functions (10-100x fewer objects), prefer column pruning (select needed columns before shuffles), and enable Kryo to shrink serialized forms. Growth fixes attack retention: unpersist() DataFrames the moment their last action completes, destroy broadcast variables past their join, and cap driver-side accumulation with bounded structures. The log shape picks the weapon — never tune blind.

Operationalize the reading. Ship executor GC summaries to the metrics store per stage (young-GC count, full-GC time, old-gen after), alert on full-GC time exceeding 10% of task time, and require a GC note on any memory-increase PR. Engineers who read GC logs size with evidence; engineers who don't buy RAM with hope. The incident's postmortem added GC summaries to every nightly job — the next skew announced itself in old-gen growth two weeks before it could kill a task.

# Executor GC logging (one executor is enough to classify)
# --conf spark.executor.extraJavaOptions='-XX:+PrintGCDetails -Xloggc:/tmp/gc.log'
# Churn signature: young GC every seconds, old-gen flat
# grep -c 'GC (Allocation Failure)' /tmp/gc.log
# Growth signature: old-gen climbs without recovery
# grep 'Full GC' /tmp/gc.log | awk '{print $NF}' | tail -n 20
# Retention hygiene in code:
# df.unpersist()  # the moment its last action completes

A Repeatable Heap-OOM Workflow

Run every heap death through victim-shape-cause in order. Victim: stage colors plus stderr location — driver corpse with green tasks is centralization, red failed tasks are concentration. Shape: task histogram and top-key share — one 41-minute giant is skew, uniform largeness is sizing, driver collect frames are result abuse. Cause: the matching evidence — schema of the hot key, broadcast sizes, GC shape, partition counts.

Fix the cause's layer only. Centralization gets bounded actions and Parquet sinks. Skew gets salting plus backfill repair. Sizing gets partition math then measured pools. Retention gets unpersist and broadcast lifecycle. Then encode: skew gates on joins, collect linters in review, GC summaries per nightly job, memory numbers with measurements attached. Fixes without gates are anecdotes; gates make them policy.

Close with cost honesty. Record DBU and instance spend before and after — the incident's rollback documented $1,400 of waste reversed, which funded the observability work. Heap incidents recur as data grows, so the workflow's product is a team that classifies in minutes, sizes from measurements, and never doubles memory fleet-wide on a histogram with one giant again.

# 1. Victim: who died?
# yarn logs -applicationId <id> | grep -B3 -A8 'OutOfMemoryError' | head -n 40
# 2. Shape: skew or size? (notebook)
# df.groupBy('uid').count().orderBy(desc('count')).limit(5).show()
# 3. Cause: partitions and peaks (Spark UI: Stages + Executors tabs)
# 4. Fix layer only: salt | bound actions | repartition | unpersist
# 5. Encode: skew gate + collect lint + GC summaries + measured pools