Last updated on 2026-07-21T20:52:21.517Z

PySpark OOM Errors on Large DataFrames: A Step-by-Step Debugging Guide

If you have ever run a Spark job that suddenly fails with an “OutOfMemoryError,” you are not alone. The pyspark out of memory error on large dataframe is one of the most common issues people face when transitioning from small- to large-scale data processing. The good news: you can fix it. This beginner-friendly, step-by-step guide shows you how to find the root cause and apply targeted fixes—without just “throwing more memory” at the problem.

In this article, you will learn:

How memory works in PySpark (driver, executors, JVM vs. Python worker, shuffle)
How to read error messages and use the Spark UI to pinpoint where memory is failing
Common operations that trigger OOM (joins, groupByKey, collect, skew, broadcast)
Concrete steps to optimize partitioning, caching, serialization, and shuffles
Practical configuration settings and code snippets that you can copy and adapt

Why PySpark Runs Out of Memory

PySpark runs on the JVM (Java Virtual Machine) and often executes user code in a separate Python worker process. That means an Out of Memory can occur in multiple places:

Driver (the “controller”): usually fails when you collect too much data to the driver or hold huge local structures.
Executor JVM: fails during shuffles, wide transformations, caching, or when tasks need more heap than available.
Python worker: fails inside UDFs or Pandas UDFs (Arrow-based) when batches do not fit in memory.

Large DataFrames can cause spikes in memory usage during these heavy operations:

Shuffles (e.g., joins, groupBy, distinct, orderBy): Data is redistributed across the cluster; temporary memory usage can exceed the size of your input.
Wide operations: groupByKey() and collect_list() can materialize very large in-memory structures.
Broadcast joins: Attempting to broadcast a large table to every executor.
Skew: A small set of keys receive a disproportionate amount of data, overloading a few tasks.
Collecting to driver: collect(), toPandas(), and take(n) with huge n quickly OOM the driver.

Recognizing OOM Symptoms and Error Messages

Here are common error messages you may encounter when you hit the pyspark out of memory error on large dataframe issue:

JVM/Executor OOM: java.lang.OutOfMemoryError: Java heap space
GC pressure: java.lang.OutOfMemoryError: GC overhead limit exceeded
Container killed (YARN/Kubernetes): Exited with exit code 137 or Container killed by YARN for exceeding memory limits
Driver OOM: Similar OutOfMemoryError but in driver logs (often triggered by collect() or toPandas())
Python worker OOM: MemoryError, worker exited, or Arrow OutOfMemoryError

Quick First Aid: Fast Ways to Avoid Common OOM Pitfalls

If your job is failing right now and you need quick relief before a deeper dive, try these safe, general fixes:

Avoid collecting large data: Replace collect() and toPandas() with write to storage or show(20, truncate=False) for small samples.
Increase partitions before wide ops: df = df.repartition(2000) (choose a number appropriate to data size and cluster).
Disable broadcast if a table is too big: spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1).
Cache to disk if caching is needed: df.persist(StorageLevel.MEMORY_AND_DISK) instead of MEMORY_ONLY.
Filter early: Push down filters and select only needed columns to reduce DataFrame width.
Use Parquet: Prefer columnar formats like Parquet/ORC over raw CSV to reduce memory and CPU.

A Step-by-Step Debugging Guide

Use this sequence to move from symptom to root cause and then to a targeted fix.

Step 1: Confirm Where the OOM Happens

Check logs to identify the failing component:

Driver OOM: The driver log shows OutOfMemoryError. Often caused by collect(), toPandas(), or building large Python lists/dicts locally.
Executor OOM: Stage tasks fail with OOM; look at the executor logs via the Spark UI (Executors tab).
Python worker: UDF stack traces or Arrow errors; you may see frequent worker restarts or MemoryError in Python.

Also check your cluster manager:

YARN: ApplicationMaster logs; containers killed due to exceeding memory limits.
Kubernetes: Pod logs and events; OOMKilled with exit code 137.

Step 2: Reproduce with a Minimal Example

Reproducing the issue in a controlled script allows safe iteration on fixes. For example, if a join is failing, create a small script focusing on that join only. Sample pattern:

from pyspark.sql import SparkSession
from pyspark.storagelevel import StorageLevel

spark = (SparkSession.builder
         .appName("oom-repro")
         .config("spark.sql.shuffle.partitions", "400")
         .getOrCreate())

left = spark.read.parquet("s3://bucket/large_left/")
right = spark.read.parquet("s3://bucket/large_right/")

# Optional: Narrow early
left = left.select("id", "feature1", "feature2")
right = right.select("id", "other_feature")

# Potential OOM: wide join + skew
joined = left.join(right, on="id", how="inner")

# Try to inspect only a sample
print(joined.limit(10).toPandas())

With a minimal script, it’s easier to tune partitions, caching, and join strategy without unrelated complexity.

Step 3: Inspect the Spark UI

Open the Spark UI for your application. Focus on:

Jobs/Stages: Which stage is failing? Are there shuffles? How many tasks?
SQL tab: View the query plan and click through to stage details.
Storage tab: See cached/persisted DataFrames, their sizes, and memory usage.
Executors tab: Check for executors with frequent task failures or high memory use.

Red flags:

A single stage with one or a few long-running tasks
Very large shuffle read/write per task
Executor lost errors concentrated on specific nodes (possible skew or resource mismatch)

Step 4: Identify the Problematic Operation

Pinpoint the transformation triggering the OOM:

Joins: Large shuffles; check size of each side and whether a broadcast is attempted.
Aggregations: groupByKey() is memory-heavy; prefer reduceByKey/aggregateByKey (RDD) or aggregations with reducers (DataFrame API).
Explodes/collect_list: Can create massive in-memory lists; consider approximate or sampled alternatives.
collect/toPandas: Driver OOM; replace with writes to disk or smaller samples.
UDFs/Pandas UDFs: Large batch sizes or unbounded state can OOM Python workers.

Step 5: Estimate Data Size and Amplification

Memory needs often exceed raw input size. During shuffles and materialization, Spark may need 2–5x temporary space. Rough checks:

Input size from storage: sum file sizes in S3/HDFS.
DataFrame width: number and types of columns (wide rows are heavy).
Skew: heavy keys concentrate memory into one or a few tasks.

Use quick metrics:

print("Partitions:", df.rdd.getNumPartitions())
print("Approx count:", df.limit(1000000).count())  # avoid counting entire huge dataset

# Estimate row width by sampling
types = {f.name: f.dataType.simpleString() for f in df.schema.fields}
print(types)

As a rule of thumb, a shuffle heavy join might need several times the in-memory footprint of both sides combined. Plan capacity accordingly, or change the algorithm to reduce intermediate size.

Step 6: Fix Partitioning and Parallelism

Too few partitions over-concentrate data (and memory) into a small number of tasks. Too many can add overhead but generally helps avoid OOMs.

Increase shuffle partitions for large joins/aggregations:

spark.conf.set("spark.sql.shuffle.partitions", 2000)  # adjust to data + cluster size

Repartition before wide operations to spread load:

df = df.repartition(2000)  # or repartitionByRange("key") for range-based shuffles

Coalesce only reduces partitions; don’t coalesce before wide shuffles.
Use partition columns to prune data when reading (e.g., dates, regions).

Step 7: Choose Safer Aggregations and Reduce Data Early

Prefer memory-efficient patterns:

Replace groupByKey() with aggregations like groupBy().agg() or key-based reducers that combine data before shuffling.
Use approx_count_distinct, bloom filters, or sampled stats when exactness is not required.
Project only needed columns with select() before joins and aggregations.
Filter early and aggressively to reduce input size.

Step 8: Tune Joins and Broadcasts

Joins are a frequent cause of the pyspark out of memory error on large dataframe. Apply these strategies:

Disable broadcasting if the “small” table is not small enough:

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

Or force broadcast only when you are sure it is small:

from pyspark.sql.functions import broadcast
joined = large_df.join(broadcast(really_small_df), "id")

Use join hints to influence strategy, e.g., merge or shuffle_hash (Spark 3):

SELECT /*+ SHUFFLE_HASH(t1) SHUFFLE_HASH(t2) */ *
FROM t1 JOIN t2 ON t1.id = t2.id

Leverage AQE (Adaptive Query Execution) for skewed joins and better partition sizes:

spark.conf.set("spark.sql.adaptive.enabled", True)
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)
spark.conf.set("spark.sql.adaptive.shuffle.targetPostShuffleInputSize", "256MB")

Fix skew: Salt high-frequency keys, or use AQE’s skew join optimization.

from pyspark.sql.functions import col, rand, concat_ws

# Salt the heavy side
salt_buckets = 10
left_salted = left.withColumn("salt", (rand()*salt_buckets).cast("int"))
right_salted = right.withColumn("salt", (rand()*salt_buckets).cast("int"))
joined = left_salted.join(right_salted, on=["id", "salt"], how="inner")

Step 9: Cache and Persist Wisely

Caching is helpful but can also cause OOM by attempting to store huge data entirely in memory.

Prefer MEMORY_AND_DISK over MEMORY_ONLY for large data:

from pyspark.storagelevel import StorageLevel
heavy_df = heavy_df.persist(StorageLevel.MEMORY_AND_DISK)
# Always unpersist when done
after = heavy_df.count()  # materialize once
heavy_df.unpersist()

Cache only what you reuse multiple times. Avoid caching transient intermediates.
Beware of caching right before a large shuffle; consider caching inputs earlier or results after reduction.

Step 10: Configure Serialization and Memory

Spark’s serializer and memory settings influence peak usage and spill behavior.

Use Kryo serializer (more compact than Java serializer):

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

Unified memory tuning (Spark 2+). If frequent spills or OOMs occur during shuffle, consider:

spark.conf.set("spark.memory.fraction", 0.6)           # default ~0.6
spark.conf.set("spark.memory.storageFraction", 0.5)    # adjust cautiously

Off-heap memory (advanced):

spark.conf.set("spark.memory.offHeap.enabled", True)
spark.conf.set("spark.memory.offHeap.size", "2g")

Note: Memory changes are not silver bullets; fix algorithms first, then tune.

Step 11: Avoid Driver OOM: Don’t Collect Big Data

Common beginner mistake: calling collect() or toPandas() on huge DataFrames. The driver is not a data warehouse.

Use limit() before collecting small samples.
Persist results to Parquet/Delta and inspect downstream with SQL or BI tools.
If you must bring data locally, stream it in chunks (e.g., using foreachPartition with care).

Step 12: Python UDFs and Pandas UDFs Memory

Python-side operations can OOM if batch sizes are large or if UDFs build large in-memory structures.

Prefer built-in Spark SQL functions over UDFs for performance and memory efficiency.
Control Arrow/Pandas UDF batch size:

spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", 5000)  # or lower

Avoid accumulating rows into Python lists or dicts inside mapPartitions or UDFs.

Step 13: Rethink Data Layout and Pruning

Schema and storage format can drive memory profile:

Use Parquet/ORC for columnar reads and predicate pushdown.
Partition your tables on frequent filters (e.g., date, country) to prune files.
Bucket or sort within partitions for frequent joins on specific keys.
Drop or cast wide columns (e.g., change double to float where acceptable).

Step 14: Skew Detection and Fixes

Skew is when a few keys have disproportionately many rows, causing one or a few tasks to OOM while others finish quickly.

Detect via Spark UI: some tasks show huge shuffle read sizes or very long durations.
Mitigate with AQE skew join, salting, or pre-aggregation by key to reduce data before the main shuffle.

Step 15: Control Task Concurrency and Resource Sizing

Even with good algorithms, you might need to balance task concurrency and memory per task.

Executors and cores: More cores per executor increase concurrent tasks, which increases memory contention.
Right-size executors: Favor more executors with fewer cores each for better isolation.
Memory overhead: Increase spark.executor.memoryOverhead if containers OOM due to off-heap/native memory.

spark = (SparkSession.builder
         .config("spark.executor.instances", 20)
         .config("spark.executor.cores", 4)
         .config("spark.executor.memory", "8g")
         .config("spark.executor.memoryOverhead", "2g")
         .config("spark.dynamicAllocation.enabled", False)
         .getOrCreate())

On managed clusters, check the cluster autoscaling and available memory; ensure you are not starved of resources.

Step 16: Manage Checkpoints and Lineage

Very long lineages (deep DAGs) can stress memory and cause repeated recomputation of heavy stages. Use checkpointing to break lineage after expensive steps:

spark.sparkContext.setCheckpointDir("s3://bucket/checkpoints/")
df = df.checkpoint(eager=True)  # materialize now, reduce lineage

Checkpointing can also reduce the chance of OOM caused by repeated recomputation due to failures.

Step 17: Avoid Gotchas That Secretly Inflate Memory

Explode: explode() on arrays with huge elements multiplies row counts; filter or limit prior to exploding.
collect_set/list: Dangerous on high-cardinality groups; consider top-k or sample instead.
toLocalIterator(): Streams data to the driver; still risky for huge results.
CSV reads: Without schema and type inference off, Spark may use more memory; always provide a schema for large CSVs.

Targeted Fixes by Error Type

java.lang.OutOfMemoryError: Java heap space

Increase partitions; reduce per-task memory pressure.
Reduce DataFrame width or filter early.
Use MEMORY_AND_DISK caching; avoid caching huge intermediates.
Consider slightly larger executor memory after fixing algorithms.

java.lang.OutOfMemoryError: GC overhead limit exceeded

Indicates most time spent in GC; reduce memory usage via partitioning and less in-memory aggregation.
Use Kryo, drop unnecessary columns, prefer safer aggregations.
Consider reducing the size of broadcast or disable it.

Exit code 137 / Container killed for exceeding memory

Increase spark.executor.memoryOverhead and/or overall executor memory.
Check Python UDFs/Arrow for large batch sizes or native memory use.
Spread work with more partitions; reduce concurrency per executor.

Driver OOM

Eliminate collect()/toPandas() on large DataFrames.
Use limit() for small samples and write results to storage to inspect.

Python MemoryError / Arrow OOM

Lower spark.sql.execution.arrow.maxRecordsPerBatch.
Avoid building large Python objects inside UDFs.
Prefer native Spark SQL functions.

Putting It All Together: A Practical Workflow

Reproduce the failing stage with a minimal script focusing on the suspect transformation (join/aggregation/UDF/explode).
Inspect Spark UI to confirm the failing stage, shuffle sizes, and skew patterns.
Reduce data early: select needed columns, push filters, use Parquet with predicate pushdown.
Right-size partitions: increase spark.sql.shuffle.partitions, and repartition before wide operations.
Stabilize joins: disable broadcast if unsafe, enable AQE and skew join handling, or salt skewed keys.
Safer aggregations: replace groupByKey and collect_list with reducers or approximations.
Cache strategically: MEMORY_AND_DISK, limit what you cache, unpersist promptly.
UDF hygiene: minimize Python-side allocations and tune Arrow batch sizes.
Only then consider adjusting executor memory/overhead and serializer if still needed.

Example: Stabilizing a Problematic Join

Suppose this job OOMs on a large inner join:

spark.conf.set("spark.sql.shuffle.partitions", 400)
left = spark.read.parquet("s3://bucket/fact/")
right = spark.read.parquet("s3://bucket/dim/")
joined = left.join(right, "id")
result = joined.groupBy("id").agg({"metric": "sum"})
result.write.mode("overwrite").parquet("s3://bucket/out/")

Step-by-step fixes:

Narrow early and disable unsafe broadcast:

left = left.select("id", "metric")
right = right.select("id", "category")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)

Increase shuffle partitions and enable AQE/skew join:

spark.conf.set("spark.sql.shuffle.partitions", 2000)
spark.conf.set("spark.sql.adaptive.enabled", True)
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)

Persist safely if reused:

from pyspark.storagelevel import StorageLevel
left = left.persist(StorageLevel.MEMORY_AND_DISK)
right = right.persist(StorageLevel.MEMORY_AND_DISK)
left.count(); right.count()  # materialize once

Execute join after repartitioning by key to spread load:

left = left.repartition(2000, "id")
right = right.repartition(2000, "id")
joined = left.join(right, "id", "inner")

Aggregate efficiently and write out:

from pyspark.sql import functions as F
result = joined.groupBy("id").agg(F.sum("metric").alias("metric_sum"))
result.write.mode("overwrite").parquet("s3://bucket/out/")

This set of changes typically reduces memory spikes and smooths out skew.

Beginner Mistakes That Cause OOM (and What To Do Instead)

Mistake: Calling df.collect() on millions of rows.
Fix: Use df.limit(1000).toPandas() for inspection; otherwise, write to storage.
Mistake: Using groupByKey() then collect_list() on high-cardinality keys.
Fix: Aggregate with reducers (groupBy().agg()), or compute top-k per group.
Mistake: Leaving default shuffle partitions (200) for very large data.
Fix: Increase spark.sql.shuffle.partitions and consider repartition by key.
Mistake: Broadcasting a not-so-small dimension table.
Fix: Disable broadcasting or ensure the broadcast side fits well below executor memory overhead.
Mistake: Caching everything with MEMORY_ONLY.
Fix: Cache selectively; prefer MEMORY_AND_DISK; always unpersist().

Useful Configuration Cheat Sheet

spark.sql.shuffle.partitions: Increase for large shuffles (e.g., 1000–4000)
spark.sql.autoBroadcastJoinThreshold: Set to -1 to disable unsafe broadcast
spark.sql.adaptive.enabled: true to enable AQE
spark.sql.adaptive.skewJoin.enabled: true to split skewed partitions
spark.serializer: org.apache.spark.serializer.KryoSerializer
spark.executor.memoryOverhead: Increase when containers die with 137
spark.memory.fraction and spark.memory.storageFraction: Tune cautiously if needed
spark.sql.execution.arrow.maxRecordsPerBatch: Lower for Pandas UDF memory control

FAQ: Fixing the PySpark Out of Memory Error on Large DataFrame

1) What causes the pyspark out of memory error on large dataframe most often?

Typically large shuffles (joins, groupBy, distinct, orderBy), broadcasting too-large tables, skewed keys, caching huge DataFrames in memory, or collecting data to the driver. Python UDFs and Arrow batch sizes can also trigger worker OOMs.

2) How do I know if the driver or executors are OOM-ing?

Check logs and the Spark UI. If you see OutOfMemoryError in the driver logs or the job crashes at a collect()/toPandas() call, it’s likely driver OOM. If tasks fail within a stage and executor logs show OOM, it’s executor-side. Exit code 137 indicates container OOM (often executor or Python worker).

3) Does increasing executor memory always help?

Not necessarily. If the algorithm is inefficient (e.g., groupByKey, huge broadcast, skew), adding memory may delay but not solve the problem. First fix the plan: prune columns, add partitions, change join strategy, and use AQE/skew handling. Then consider modest memory increases.

4) What’s a safe way to cache large DataFrames?

Use MEMORY_AND_DISK, materialize once (e.g., count()), and unpersist when done. Cache only reused DataFrames and verify cache size in the Spark UI (Storage tab).

5) How do I deal with skewed keys that cause OOM?

Enable AQE skew join, salt heavy keys, and pre-aggregate data before the main join. Inspect skew using the Spark UI; tasks reading hundreds of MB/GB while others read much less is a skew signal.

6) Is disabling broadcast a good idea?

Disable auto-broadcast if your “small” table is not truly small or is unpredictable after filters. Otherwise, broadcasting genuinely small dimension tables can improve performance.

7) Why do Pandas UDFs crash with out of memory?

Pandas UDFs load batches into Python memory using Arrow. If batches are too large or the function builds large temporary structures, workers can OOM. Reduce batch size and avoid heavy Python-side accumulation.

8) Can I fix OOM by switching to Parquet?

Often yes. Parquet is columnar and supports predicate and column pruning, reducing memory and I/O significantly versus CSV. Always provide a schema when reading large CSV to avoid expensive inference.

9) How many shuffle partitions should I use?

It depends on data size and cluster. Start with a few thousand (e.g., 1000–4000) for very large jobs and adjust based on observed shuffle sizes and task durations. More partitions generally lowers per-task memory pressure.

10) What’s the difference between MEMORY_ONLY and MEMORY_AND_DISK?

MEMORY_ONLY keeps cached data only in RAM and will evict partitions when memory is tight, possibly causing recomputation and OOM. MEMORY_AND_DISK spills cached partitions to disk when RAM is insufficient, greatly reducing OOM risk at a small performance cost.

Checklist: Before You Re-Run

Removed or limited collect()/toPandas() usage
Selected only required columns and pushed filters early
Increased spark.sql.shuffle.partitions and repartitioned by key if needed
Enabled spark.sql.adaptive.enabled and skew join handling
Disabled auto-broadcast when unsafe, or explicitly broadcast only truly small tables
Replaced groupByKey/heavy collect_list with safer aggregations
Switched caching to MEMORY_AND_DISK and unpersisted when done
Tuned Pandas UDF batch sizes or replaced with built-in functions
Optionally enabled Kryo serializer, and adjusted memory/overhead if still necessary

Conclusion

The pyspark out of memory error on large dataframe isn’t just a hardware problem—it’s often a planning and design issue. By understanding where memory pressure occurs (driver, executors, Python workers, and during shuffles), you can pinpoint the failing operation and apply targeted fixes: prune data early, increase and align partitions, choose safer aggregations, stabilize joins with AQE and skew handling, and cache to disk when needed. Only after these algorithmic improvements should you adjust memory settings. With this step-by-step approach, you can turn frustrating OOM failures into stable, scalable PySpark pipelines.