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

> Fix common out-of-memory errors in Apache Spark. Learn to prevent JVM heap exhaustion with optimized memory allocation, avoiding collect and tuning caching and serialization.

- Repository: [DataTalksClub/data-engineering-zoomcamp](https://github.com/DataTalksClub/data-engineering-zoomcamp)
- Tags: how-to-guide
- Published: 2026-05-30

---

**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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/06-batch/code/06_spark_sql.py)—you must explicitly configure driver memory based on your data volume.

Increase the driver heap size using:

```python
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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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:

```python
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:

```python
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**:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/streaming_confluent.py) and [`07-streaming/extras/python/streams-example/pyspark/streaming.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/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:

```python
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.