Common Causes of Out-of-Memory Errors in Apache Spark and How to Prevent Them

Out-of-memory errors in Apache Spark occur when the driver or executor JVMs exhaust heap space during distributed computations, and can be prevented by properly sizing memory allocations, avoiding unbounded collect() operations, and tuning serialization and caching parameters.

Out-of-memory (OOM) errors are among the most common failure modes in Apache Spark applications, particularly when processing large-scale datasets in the DataTalksClub/data-engineering-zoomcamp repository workflows. These errors manifest in either the driver process—which maintains the Spark context and aggregates results—or the executor processes that perform the actual distributed computation. Understanding the specific root causes and implementing targeted prevention strategies is essential for building robust data pipelines.

Driver-Side Out-of-Memory Errors

The Spark driver maintains the logical execution plan, broadcasts variables to executors, and collects final results. When the driver JVM cannot allocate memory for these operations, the entire application fails.

Insufficient Driver Memory Configuration

The driver crashes when spark.driver.memory is too low to hold the Spark SQL logical plan, broadcast variables, and application metadata. In production workloads—unlike the minimal SparkSession shown in 06-batch/code/06_spark_sql.py—you must explicitly configure driver memory based on your data volume.

Increase the driver heap size using:

spark = (SparkSession.builder
         .appName("oom-prevention-demo")
         .config("spark.driver.memory", "4g")
         .getOrCreate())

Large Result Collections

Calling collect() or toPandas() on large DataFrames forces the driver to materialize the entire distributed dataset into a single JVM. This is the most frequent cause of driver OOM.

Prevention: Write results to external storage (GCS, S3, BigQuery) instead of collecting to the driver. Reference 06-batch/code/06_spark_sql_big_query.py for patterns that write large outputs directly to BigQuery without intermediate materialization.

Executor Out-of-Memory Errors

Executors perform the actual data processing and shuffling. OOM occurs when task data, shuffle buffers, and cached DataFrames exceed the allocated heap.

Inadequate Executor Memory Allocation

Each executor requires sufficient RAM to hold task partitions and temporary shuffle data. Small spark.executor.memory settings cause immediate failure during wide transformations like groupBy() or join().

Configure executor memory and tune the memory fractions:

spark = (SparkSession.builder
         .config("spark.executor.memory", "8g")
         .config("spark.memory.fraction", "0.8")
         .config("spark.memory.storageFraction", "0.5")
         .getOrCreate())

The spark.memory.fraction (default 0.6) controls the fraction of heap used for Spark execution and storage, while spark.memory.storageFraction (default 0.5) reserves portion of that space for cached data.

Uncontrolled DataFrame Caching

Persisting large DataFrames with MEMORY_ONLY forces Spark to retain all data in RAM. When cached size exceeds available executor memory, the JVM crashes.

Use MEMORY_AND_DISK or DISK_ONLY storage levels for large tables:

from pyspark.storagelevel import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)

# ... transformations ...

df.unpersist()  # Explicitly release memory

Join and Shuffle Operations

Oversized Broadcast Joins

Spark automatically broadcasts the smaller side of a join to all executors when it falls below spark.sql.autoBroadcastJoinThreshold (default 10MB). If you force broadcast on large DataFrames or the threshold is misconfigured, each executor OOMs when receiving the oversized table.

Ensure broadcast tables stay under 100MB:

from pyspark.sql.functions import broadcast

small_df = spark.read.parquet("s3://bucket/dimensions/")
large_df = spark.read.parquet("s3://bucket/facts/")

if small_df.rdd.map(lambda r: len(str(r))).sum() < 100 * 1024 * 1024:
    result = large_df.join(broadcast(small_df), "id")
else:
    result = large_df.join(small_df, "id")  # Shuffle join

Disable automatic broadcast joins entirely when working with unknown table sizes:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

Data Skew and Expensive Shuffles

When a key is highly skewed, a single reducer task receives disproportionate data during shuffles, exhausting its memory during the merge phase. This is evident in long-running tasks in the Spark UI.

Mitigate skew by salting or repartitioning:

balanced = df.repartition(200, "skewed_key")
balanced.groupBy("skewed_key").count().show()

Increase shuffle parallelism via spark.sql.shuffle.partitions to distribute load across more tasks.

Serialization and Python UDF Overhead

Partition Management Overhead

Over-partitioning creates thousands of small tasks, each consuming JVM overhead for task metadata. The driver OOMs trying to manage the task queue.

Coalesce to a sensible partition count:

df.coalesce(100)  # Reduce partitions before writing

Set spark.default.parallelism based on your cluster core count.

Python UDF Memory Serialization

Python UDFs (including pandas_udf) serialize data between the JVM and Python workers using Py4J. Large payloads overflow the JVM heap during this exchange.

Prefer built-in Spark SQL functions over UDFs. When UDFs are unavoidable, enable Arrow serialization for zero-copy transfers:

spark.conf.set("spark.sql.execution.arrow.enabled", "true")

@F.pandas_udf("double")
def add_one(s: pd.Series) -> pd.Series:
    return s + 1

Inefficient Java Serialization

The default Java serializer creates bulky byte arrays for complex objects, consuming more heap than expected during shuffles.

Switch to Kryo serialization for compact binary representation:

spark = (SparkSession.builder
         .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
         .getOrCreate())

Implementation Patterns from the Data Engineering Zoomcamp

The Zoomcamp repository demonstrates these principles across batch and streaming workloads. In cohorts/2023/week_6_stream_processing/streaming_confluent.py and 07-streaming/extras/python/streams-example/pyspark/streaming.py, streaming applications maintain state and watermarks in memory—requiring careful tuning of spark.streaming.backpressure.enabled and checkpoint intervals to prevent OOM in long-running jobs.

Complete configuration template for production workloads:

from pyspark.sql import SparkSession

spark = (SparkSession.builder
         .appName("zoomcamp-oom-safe")
         .config("spark.driver.memory", "4g")
         .config("spark.executor.memory", "8g")
         .config("spark.memory.fraction", "0.75")
         .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
         .config("spark.sql.execution.arrow.enabled", "true")
         .config("spark.sql.autoBroadcastJoinThreshold", "50m")
         .config("spark.sql.shuffle.partitions", "200")
         .getOrCreate())

Summary

• Size memory appropriately: Configure spark.driver.memory and spark.executor.memory based on actual data volumes, not default values.
• Avoid driver collections: Never call collect() on large datasets; write to external storage instead.
• Cache wisely: Use MEMORY_AND_DISK storage levels and explicitly unpersist() when done.
• Control broadcast thresholds: Ensure broadcast joins stay under 100MB or disable automatic broadcasting.
• Balance partitions: Repartition skewed data and coalesce over-partitioned datasets.
• Optimize serialization: Enable Kryo for Java and Arrow for Python UDFs.
• Monitor continuously: Watch the Spark UI Storage and Executors tabs for memory pressure indicators.

Frequently Asked Questions

What is the most common cause of out-of-memory errors in Spark?

The most common cause is calling collect() or toPandas() on large DataFrames, which forces the driver to materialize distributed data into a single JVM heap. This also occurs when broadcasting large DataFrames in joins or when executors lack sufficient memory to handle skewed shuffle operations.

How do I know if an OOM error occurred in the driver or an executor?

Driver OOMs appear as OutOfMemoryError in the driver logs with stack traces mentioning collect() or task scheduling. Executor OOMs appear in individual executor logs with errors during shuffle merges or cached R eviction. The Spark UI shows failed tasks with "Container killed by YARN/Mesos for exceeding memory limits" for executor issues.

What is the difference between MEMORY_ONLY and MEMORY_AND_DISK persistence?

MEMORY_ONLY stores DataFrames exclusively in RAM and re-computes partitions from lineage if memory is exhausted, which can cause recomputation overhead. MEMORY_AND_DISK spills overflow partitions to local disk, preventing OOM at the cost of slower disk I/O. Use MEMORY_AND_DISK when total cached data might exceed available executor memory.

How does data skew cause out-of-memory errors?

Data skew occurs when one or few keys contain the majority of records. During shuffle operations (like groupByKey or join), all records for a skewed key route to a single executor task. This executor receives disproportionate memory pressure and OOMs while other executors remain underutilized. Fix this by salting keys or repartitioning with a higher partition count.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}