Why am I getting an OutOfMemoryError in a Spark executor?

Summary

• One cause the Spark tuning guide singles out: a single task's working set is too large, because shuffles (groupByKey, reduceByKey, join) build a hash table inside each task.
• Its simplest fix is more parallelism so each task's input is smaller, at 2-3 tasks per CPU core.
• Java objects can take a factor of two to five times the space of the raw data, so use primitive arrays and, in Scala or Java, serialized caching with Kryo.
• Spark recommends leaving spark.memory.fraction (0.6) and spark.memory.storageFraction (0.5) at their defaults.
• The Spark UI shows where memory goes. Databricks suggests testing with double the memory per core.

Why am I getting an OutOfMemoryError in a Spark executor?

The Spark tuning guide describes one cause of this error: the RDDs are not too big for memory, but the working set of a single task is. Shuffle operations (sortByKey, groupByKey, reduceByKey, join) build a hash table within each task to perform the grouping, and that hash table can be large.

How do I tell what is running out of memory?

Symptom	Likely cause	Where to look
Failure in a shuffle stage	One task builds a hash table that is too large	Stages tab: peak execution memory, shuffle spill
Heap use is far above the size of the raw data	Java objects can take a factor of two to five times the space of the raw data	Storage tab: cache a sample and read its size
Long garbage collection time	GC cost is proportional to the number of Java objects	Executors tab: GC time
"Memory Overhead Exceeded" errors, which the docs say non-JVM tasks commonly hit	Non-heap memory is exhausted	spark.executor.memoryOverhead
A few tasks fail while others finish	Skewed partitions	Stages tab; adaptive query execution splits skewed partitions in sort-merge joins

How do I fix a shuffle-related OutOfMemoryError?

Increase the level of parallelism so each task's input is smaller. The guide recommends 2-3 tasks per CPU core, and says going above the core count is safe because Spark reuses one executor JVM across many tasks. In the RDD API, pass the partition count to the shuffle. For a sum or average per key, the PySpark docs say reduceByKey or aggregateByKey performs much better than groupByKey.

# pairs: an RDD of (key, value) tuples; 200 is an example
grouped = pairs.groupByKey(numPartitions=200)

spark.default.parallelism sets the default partition count for RDDs returned by transformations such as join and reduceByKey. DataFrames and Spark SQL use spark.sql.shuffle.partitions (default 200). Set either at launch with --conf, which overrides spark-defaults.conf but not values set on SparkConf:

./bin/spark-submit --conf spark.default.parallelism=400 --conf spark.sql.shuffle.partitions=400 myApp.jar

With adaptive query execution (spark.sql.adaptive.enabled, true by default), Spark coalesces small shuffle partitions and handles skew in sort-merge joins by splitting skewed partitions.

How do I reduce the memory each record needs?

A Java object header is about 16 bytes, and a Java String adds about 40 bytes of overhead and stores each character as two bytes. The tuning guide suggests:

• Prefer arrays of objects and primitive types over collection classes such as HashMap.
• Avoid nested structures with many small objects and pointers.
• Use numeric IDs or enumeration objects instead of strings for keys.
• In Scala or Java, persist with MEMORY_ONLY_SER, which keeps each partition as one large byte array at the cost of slower access, and use Kryo (--conf spark.serializer=org.apache.spark.serializer.KryoSerializer).

In Python, stored objects are always serialized with Pickle, so a serialized storage level makes no difference, and MEMORY_ONLY_SER is not in the guide's list of Python storage levels.

How do I size executor memory and check garbage collection?

If you cannot add partitions, give each executor more memory. The defaults are 1g for spark.executor.memory, and for spark.executor.memoryOverhead the executor memory times spark.executor.memoryOverheadFactor (0.10), with a floor of spark.executor.minMemoryOverhead (384m). On YARN and Kubernetes, an executor's maximum container memory is the sum of spark.executor.memoryOverhead, spark.executor.memory, spark.memory.offHeap.size and spark.executor.pyspark.memory.
Inside the heap, execution and storage share a unified region sized by spark.memory.fraction (default 0.6 of the heap minus 300MiB). spark.memory.storageFraction (default 0.5) protects cached blocks from eviction. Execution can evict storage down to that threshold, but storage cannot evict execution. The guide says the typical user should not need to adjust either setting.
To measure garbage collection, pass the guide's flags through the executor Java options. The logs land in the stdout files in the executors' work directories, not on the driver. With less than 32 GiB of RAM, the guide also suggests -XX:+UseCompressedOops.

./bin/spark-submit --conf "spark.executor.extraJavaOptions=-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps" myApp.jar

What is different on Databricks?

The Databricks Spark memory issues guide lists these typical causes: too few shuffle partitions, a large broadcast, UDFs, a window function without PARTITION BY, skew and streaming state. To confirm memory is the problem, double the memory per core, for example by moving from a worker type with 4 cores and 16GB to one with 4 cores and 32GB. If the job takes longer to fail or stops failing, that is a good sign memory is the cause. Databricks also advises against hardcoding settings such as spark.executor.memory in compute or job definitions. Photon runs natively in C++ instead of the JVM, which the docs say eliminates garbage collection pauses and memory overhead, but it does not support UDFs, RDD APIs or Dataset APIs.

FAQs

Should I always use Kryo?

The guide describes Kryo as a significantly faster, more compact alternative to Java serialization, but it does not support all Serializable types and needs classes registered in advance for best performance. That requirement is the only reason it is not the default. Unregistered classes still work, but Kryo stores the full class name with each object.

How do I monitor memory while a job runs?

The Executors tab shows each executor's memory, disk and core usage, storage memory reserved for cached data and GC time. The Storage tab lists persisted RDDs and DataFrames after an action has materialized them.