Executor memory: failures and fixes

Symptom

java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: GC overhead limit exceeded
Job aborted due to stage failure: Task 37 in stage 4.0 failed 4 times

Executors die, the stage retries, and after four failed attempts the job aborts.

Why it happens

One task needs more heap than its share (about pool ÷ cores) for something that can't spill:

• Partitions far too large, for example 500 GB through the default 200 shuffle partitions is about 2.5 GB each.
• A skewed key sends most rows to one task.
• Unbounded groups: collect_list, collect_set or groupByKey on a key with millions of rows.
• explode or a many-to-many join multiplying rows unexpectedly.
• Very wide rows: large strings, JSON blobs, binary columns.

How to confirm

1. Stages tab → failed tasks → read the error column.
2. Only a few tasks fail? Compare max vs median Shuffle Read Size: that's skew.
3. All tasks fail? Divide shuffle read by task count. Over ~1 GB per task means partitions are too big.
4. SQL tab: look for a sudden jump in rows output after an explode or join.

Fix

• Smaller partitions: raise spark.sql.shuffle.partitions, or lower AQE's advisory size and let it split.
• Fix skew: AQE skew join, broadcast the small side, salt the hot key, filter NULL keys.
• More memory per task: fewer cores per executor, or more executor memory.
• Replace unbounded collect_list with a real aggregation, or cap it.
• Drop unneeded wide columns before the shuffle.

spark.conf.set("spark.sql.shuffle.partitions", 2000)              # ~250 MB each for 500 GB
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")
# at cluster launch, if all else is tuned:
# spark.executor.cores=4   spark.executor.memory=24g

Prevent

• Aim for 128–200 MB per shuffle partition; keep AQE on; watch max-vs-median task metrics on large jobs.

Symptom

Container killed by YARN for exceeding memory limits.
18.1 GB of 18 GB physical memory used.
Consider boosting spark.executor.memoryOverhead.

# Kubernetes: executor pod status OOMKilled, exit code 137

The executor disappears with no OutOfMemoryError in its own log. The heap was fine; the process as a whole went over the container limit, so the cluster manager killed it.

Why it happens

• Python workers for PySpark UDFs and pandas UDFs, one per running task, each with its own memory.
• Arrow buffers moving data between the JVM and Python.
• Native and off-heap memory: Netty shuffle buffers, compression libraries, Parquet/ORC readers.
• Overhead left at the default 10%, which is sized for JVM-only jobs.

How to confirm

1. The message comes from YARN or Kubernetes, not from the JVM.
2. Executors tab: peak JVM heap used is well below spark.executor.memory.
3. The failing stage runs Python UDFs, applyInPandas or very large shuffles.

Fix

• Raise spark.executor.memoryOverhead, often to 15–25% of heap for PySpark-heavy jobs.
• Set spark.executor.pyspark.memory to give Python workers an explicit budget, added to the container.
• Fewer cores per executor means fewer Python workers running at once.
• Replace Python UDFs with built-in functions; use pandas UDFs over plain Python UDFs.
• Smaller Arrow batches: spark.sql.execution.arrow.maxRecordsPerBatch (10,000 by default).

spark.executor.memory=16g
spark.executor.memoryOverhead=4g        # instead of the default 1.6g
spark.executor.pyspark.memory=2g        # optional explicit Python budget

Prevent

• Size overhead by workload type (JVM-only vs PySpark-heavy); prefer built-ins; keep a PySpark cluster profile with higher overhead.

Symptom

java.lang.OutOfMemoryError: Java heap space        (in the driver log)
Total size of serialized results of 412 tasks (1024.0 MiB)
is bigger than spark.driver.maxResultSize (1024.0 MiB)

The whole application dies, or a notebook detaches. Executors were healthy.

Why it happens

• collect(), toPandas() or take(n) with a large n on big data.
• Building a broadcast: by default the driver collects the small side first.
• Planning over hundreds of thousands of files or partitions.
• Very long query plans built in loops (each iteration extends the lineage).

How to confirm

1. Find the last action in the driver log before the crash.
2. Look for collect / toPandas / broadcast() on large data in that code path.
3. Check file counts on the input tables.

Fix

• Write results to a table or files instead of collecting; use limit() for samples.
• Lower the broadcast threshold or remove a broadcast hint on a large table.
• Compact small files; in loops, write intermediate results or checkpoint() to cut lineage.
• Only then: raise spark.driver.memory and spark.driver.maxResultSize (default 1g).

# instead of
pdf = big_df.toPandas()
# do
big_df.write.mode("overwrite").saveAsTable("tmp.result")
sample = big_df.limit(1000).toPandas()

Prevent

• Code-review rule: no collect or toPandas on unbounded data.

Symptom

Not enough memory to build and broadcast the table to all worker nodes.
As a workaround, you can either disable broadcast by setting
spark.sql.autoBroadcastJoinThreshold to -1 or increase the spark driver memory
by setting spark.driver.memory to a higher value.

Could not execute broadcast in 300 secs.
Cannot broadcast the table that is larger than 8GB.

Why it happens

• The threshold was raised too high, or a broadcast() hint was put on a table that grew.
• Wrong size estimates: missing or stale statistics make a large table look small.
• On-disk size isn't memory size. Compressed Parquet often expands 5–10× in memory, so a 300 MB table can need 2–3 GB.
• Every executor holds a full copy, so a big broadcast also squeezes executor memory.

How to confirm

1. explain() shows BroadcastExchange on the table you expected to be large.
2. SQL tab: the broadcast node's data size and build time.

Fix

• Lower spark.sql.autoBroadcastJoinThreshold or remove the hint; let sort-merge join handle it.
• Refresh statistics: ANALYZE TABLE … COMPUTE STATISTICS.
• Rely on AQE to choose at runtime, using real sizes after filters.
• If it's slow rather than too big, raise spark.sql.broadcastTimeout (300 s).

Prevent

• Keep broadcasts to a few hundred MB in memory; don't blanket-raise the threshold cluster-wide.

Symptom

The stage is slow. Its metrics show Spill (Memory) and Spill (Disk), often several times the shuffle read size.

Why it happens

A sort, join or aggregation needed more execution memory than its task had, so Spark wrote sorted runs to local disk and merged them later. Spill (Memory) is the in-memory size of what was spilled; Spill (Disk) is its serialized size on disk.

How to confirm

1. Shuffle read ÷ tasks gives data per task. Far above 200 MB means partitions are too big.
2. Spill spread evenly across tasks means partitions are too big; spill in a few tasks means skew.

Fix

• More shuffle partitions, or a smaller AQE advisory size.
• Fewer cores per executor, or memory-optimised instances.
• Select fewer columns before the shuffle; aggregate before exploding.
• Fast local disks (NVMe) for shuffle-heavy clusters.

The full walkthrough with example metrics is in Module 4: Disk spill scenario →

Symptom

ExecutorLostFailure (executor 7 exited caused by one of the running tasks)
Reason: Executor heartbeat timed out after 120000 ms

Tasks are slow and erratic; GC time in the Executors tab is a large share of task time (over about 10% is a warning).

Why it happens

• Very large heaps full of objects cause long stop-the-world collections, and the executor misses its heartbeats.
• Many cores sharing one heap create objects faster than GC clears them.
• Caching large deserialized datasets fills the heap with long-lived objects.
• Sometimes the node really is gone, for example a spot instance reclaimed.

How to confirm

1. Executors tab: GC time against task time.
2. Executor logs: GC log lines around the timeout.
3. Cloud console: was the node pre-empted or terminated?

Fix

• Smaller executors: around 4–5 cores and a moderate heap (often 16–32 GB) rather than one giant executor per node.
• Cache less, or cache serialized (MEMORY_AND_DISK_SER), and unpersist when done.
• Fix skew so no single task builds huge in-memory structures.
• Raise spark.network.timeout only as a stopgap; it hides the problem.
• For spot loss: graceful decommissioning (deep dive) and mixed on-demand/spot pools.

Symptom

org.apache.spark.SparkException: Python worker exited unexpectedly (crashed)
# or the container is killed while a groupBy().applyInPandas() stage runs

Why it happens

groupBy().applyInPandas() loads all rows of one group into the Python worker as a single pandas DataFrame before calling your function. One huge group, often a skewed key, exhausts Python memory. pandas operations also make copies, so peak memory can be several times the data size.

How to confirm

1. Count rows per group key: df.groupBy("key").count().orderBy(F.desc("count")).
2. The failure happens in the stage running the pandas function, and the largest group dwarfs the median.

Fix

• Make groups smaller: group by a finer key, or salt the key and combine the results.
• Select only the columns the function needs, before the group.
• Use built-in aggregations where possible; use mapInPandas for streaming over batches when you don't need whole groups.
• Give Python more room: spark.executor.pyspark.memory, more overhead, fewer cores.

Symptom

After adding .cache(), joins get slower and spill more. The Storage tab shows a Fraction Cached below 100%, and parts of the cached data are recomputed anyway.

Why it happens

• Cache lives in the same unified pool as execution. Up to half the pool is protected for cache, so execution has less room.
• With MEMORY_ONLY, partitions that don't fit are dropped and recomputed from lineage each time they're used.
• Cached data that's never unpersisted keeps holding memory for the rest of the application.

Fix

• Cache only DataFrames that are reused by several actions and expensive to recompute.
• Cache after filtering and selecting columns, not the raw input.
• Use MEMORY_AND_DISK (the DataFrame default) and call unpersist() when done.
• For long pipelines, write an intermediate table instead of caching.