Why am I getting executor OutOfMemoryError?

Summary

• An executor OutOfMemoryError doesn't necessarily mean your dataset is too big for the cluster. The Spark tuning guide's example is a single task's working set, such as a reduce task in groupByKey, whose per-task hash table gets too large.
• The simplest documented fix is more parallelism, so each task gets less data: raise spark.sql.shuffle.partitions (default 200) and keep adaptive query execution on, since it splits skewed join partitions.
• Other causes include a broadcast side that's bigger than expected (spark.sql.autoBroadcastJoinThreshold, default 10MB), heavy caching, and Python or native memory outgrowing spark.executor.memoryOverhead.
• Set spark.executor.memory and memory overhead at launch, with spark-submit or the cluster configuration. Deploy-related properties may not take effect when set from a running session.

Why am I getting executor OutOfMemoryError?

The JVM throws OutOfMemoryError when it can't allocate an object and the garbage collector can't free enough memory. On a Spark executor, the cause isn't necessarily the size of your dataset. The Spark tuning guide points out that you can get an OutOfMemoryError because the working set of a single task, such as a reduce task in groupByKey, was too large: shuffle operations like groupByKey, reduceByKey, sortByKey and join build a hash table within each task. Its simplest fix is to increase the level of parallelism so each task's input is smaller. Check partitioning before you add memory.

How is executor memory divided?

Each executor is a JVM whose heap is set by spark.executor.memory (default 1g). Inside the heap, spark.memory.fraction (default 0.6) of the heap minus 300 MiB forms a unified region shared by execution memory (shuffles, joins, sorts, aggregations) and storage memory (cached data). Execution can evict cached blocks down to the protected share set by spark.memory.storageFraction (default 0.5), but storage can't evict execution. The rest of the heap holds user data structures, internal metadata and a safeguard against OOM errors from unusually large records. The docs recommend leaving both fractions at their defaults.
Outside the heap, spark.executor.memoryOverhead covers VM overheads, interned strings and other native memory. It defaults to executor memory times spark.executor.memoryOverheadFactor (0.10, or 0.40 for Kubernetes non-JVM jobs), with a minimum of 384 MiB, and applies on YARN and Kubernetes. The container's limit is the sum of spark.executor.memoryOverhead, spark.executor.memory, spark.memory.offHeap.size and spark.executor.pyspark.memory. If spark.executor.pyspark.memory isn't set, Spark doesn't limit Python's memory use, and Python workers share the overhead space with other non-JVM processes.

Which causes and fixes should you check?

Symptom	Likely cause	Fix
OOM in a shuffle stage, with a few tasks reading far bigger shuffle blocks	Too few or skewed shuffle partitions	Raise spark.sql.shuffle.partitions; keep spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled on
OOM during a join that the plan shows as a broadcast hash join	The broadcast side is bigger than expected	Lower spark.sql.autoBroadcastJoinThreshold, or set it to -1 to disable automatic broadcasts
High storage memory in the Executors tab	Large cached DataFrames	Cache only what you reuse, and release it with df.unpersist()
"Memory Overhead Exceeded" errors, often with Python UDFs	Python or native memory outside the heap	Raise spark.executor.memoryOverhead, or cap Python with spark.executor.pyspark.memory
OOM while a stream catches up on a large backlog	Each micro-batch takes in too many new files	Cap it with the file source option maxFilesPerTrigger or maxBytesPerTrigger (one, not both)

Runtime SQL settings, such as shuffle partitions and the broadcast threshold, can change per session:

spark.conf.set("spark.sql.shuffle.partitions", "400")   # default is 200
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")  # disable auto-broadcast
df.explain(mode="formatted")  # confirm the join strategy

With adaptive query execution on (the default since Spark 3.2.0), Spark also coalesces small post-shuffle partitions and, when spark.sql.adaptive.skewJoin.enabled is true, splits skewed sort-merge join tasks into roughly even ones.
The configuration guide warns that deploy-related properties may not take effect when set programmatically at runtime, so pass them at launch:

./bin/spark-submit \
  --executor-memory 8g \
  --conf spark.executor.memoryOverhead=2g \
  --conf spark.executor.pyspark.memory=2g \
  app.py

How do you find the task that ran out of memory?

Open the failed stage in the Spark UI. Its task table breaks out per-task metrics, including peak execution memory (the memory used by data structures created during shuffles, aggregations and joins), shuffle read size, shuffle spill to memory and disk, and GC time. A handful of tasks with outsized shuffle reads points to skew. The Executors tab shows each executor's memory, disk, storage memory reserved for cached data and GC time, and the Storage tab lists persisted RDDs and DataFrames.

How Databricks approaches this

On Databricks classic compute, put memory properties in the compute's Spark config field, one key-value pair per line, and pick an instance type family suited to memory-intensive workloads. Serverless compute doesn't expose executor memory: it supports only a short list of Spark properties, including spark.sql.shuffle.partitions (default auto), and caps user-defined code such as UDFs, map and mapPartitions at 1 GB of memory. For large streaming sources on serverless, Databricks recommends setting maxFilesPerTrigger or maxBytesPerTrigger to prevent out-of-memory errors.

FAQs

Should I just increase spark.executor.memory?

Not first. The tuning guide's simplest fix for reduce-task OOM is more parallelism. Add executor memory when a single task legitimately needs a larger working set, and set it at launch.

Why does the driver run out of memory while executors are fine?

collect() fetches the entire dataset to the driver, which can make the driver run out of memory; use take() when you only need a few rows. spark.driver.maxResultSize (default 1g) aborts jobs whose serialized results exceed the limit, which protects the driver.

Can caching cause executor OOM?

Cached data shares the unified region with execution, but execution can evict cached blocks above the protected share. df.cache() defaults to the MEMORY_AND_DISK_DESER storage level, which keeps partitions that don't fit in memory on disk. Unpersist DataFrames you no longer need.