Home / Data Platform / Databricks & Spark

Data PlatformTroubleshootingAdvanced

Sev1: critical Spark job keeps failing — ExecutorLostFailure, OOM and FetchFailedException

Get a failing production Spark job through tonight's run. Read the failure in the Spark UI, separate driver OOM, executor OOM, container kills and shuffle fetch failures, apply the quick fix and plan the real one.

Deepak Bhardwaj2 min read

Applies toApache Spark 3.3+ on Databricks, Synapse, EMR, HDInsight and Kubernetes

Symptoms

  • ExecutorLostFailure (executor 12 exited caused by one of the running tasks) Reason: Container killed ... exceeding memory limits / exit code 137.
  • java.lang.OutOfMemoryError: Java heap space or GC overhead limit exceeded.
  • org.apache.spark.shuffle.FetchFailedException and stage retries.
  • Total size of serialized results ... is bigger than spark.driver.maxResultSize.

1. Find where it fails

Spark UI → Stages → the failed stage. Look at:

  • Task duration and input size quartiles: max far above median = skew.
  • Shuffle read/write size and spill (memory/disk): heavy spill = partitions too large.
  • The failure reason in the failed task's row.

2. Map the error to a fix

Error Usually means Quick fix
Driver OOM, maxResultSize collect(), toPandas(), huge broadcast Don't collect; write to storage; lower broadcast threshold
Executor OOM / exit 137 Partitions too big, skewed key, wide rows, UDF memory More shuffle partitions, enable AQE skew join, bigger executors
FetchFailedException Executor holding shuffle files died (often OOM or spot loss) Fix the OOM; avoid spot for shuffle-heavy stages
Python worker killed pandas UDF / Python memory outside JVM Raise memory overhead, smaller batches (arrow.maxRecordsPerBatch)

3. Quick configuration fixes

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.shuffle.partitions", "800")          # or "auto" on Databricks
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(64 * 1024 * 1024))

Executor-level settings (set at cluster/session start, not at runtime):

spark.executor.memory 16g
spark.executor.memoryOverhead 4g       # more for PySpark / native libraries
spark.executor.cores 4

Fewer cores per executor gives each task more memory.

Tip

On Databricks, switching to a memory-optimised node type with the same total cores is often the fastest way to get tonight's run through. Then fix the code tomorrow.

4. Fix skew in code

Find the hot key:

df.groupBy("customer_id").count().orderBy("count", ascending=False).show(10)

Options: filter or handle NULL/default keys separately (a classic hot key), broadcast the small side, or salt the key:

from pyspark.sql import functions as F
N = 16
big = big.withColumn("salt", (F.rand() * N).cast("int"))
small = small.withColumn("salt", F.explode(F.array([F.lit(i) for i in range(N)])))
joined = big.join(small, ["customer_id", "salt"])

5. Re-run safely

Make sure the job is idempotent (overwrite partition or MERGE) before re-running, so a partial previous attempt doesn't leave duplicates.

Prevent it next time

  • Track stage spill and max task time per run; alert when they grow.
  • Load test with production-size data; data growth is what usually tips a job over.
  • Avoid collect() in production code paths; review UDFs for memory use.

Related fixes