Introduction

Apache Spark is built to efficiently manage large-scale data processing. One of the key reasons it performs so well is its ability to process data in memory rather than relying heavily on disk I/O. This advantage also brings new challenges. In practice, Spark jobs rarely fail because the CPU is too slow. Instead, they fail because memory becomes a bottleneck.

java.lang.OutOfMemoryError: Java heap space

A common situation is when a job runs perfectly fine on a small dataset, but once the input grows, performance drops much more than you would expect. Processing more data will naturally take longer, but sometimes the slowdown is much larger than the increase in data. 

Most of the time, this is not because you “don’t have enough RAM”, but because of how Spark uses the RAM you’ve assigned to it.

In this blog, we will focus on Spark memory and the problems it faces. So first, we will construct a simple mental model of executors, partitions, and shuffles, which are considered as the three pillars of Spark performance. We’ll then learn more about how Spark splits executor memory internally, and how to avoid crashing traps. Finally, we’ll try examples of memory optimization in action.

 

Executors: The Brains of Spark

Executors are the processes that actually do the work in Spark. Every executor is a JVM process running on a worker node in your cluster. They’re responsible for running tasks, caching data, and sending results back to the driver.

When you configure Spark executor using:

--executor-memory 8G
--executor-cores 4

you are essentially telling it to:

  • “Provide each executor 8 GB of heap memory.”
  • “Allow it to run 4 tasks in parallel.”

It’s important to understand what this means in practice: the 8 GB of memory is allocated to the executor itself, so all four tasks running on it must share that same heap; it’s 8 GB total, not 8 GB per task. If a task requires 6 GB, it can overwhelm the entire executor, even if the other three tasks aren’t pulling their weight.

This is why the common approach to speeding up a Spark job by simply adding more cores doesn’t necessarily improve performance. If the executor’s memory stays the same while the number of cores increases, you’re just running more tasks within the same memory space. They end up fighting over the heap. 

Executor tuning is about finding a good balance. You want enough cores to use the CPU well. You also want enough memory so tasks don’t feel cramped. And the workload still has to fit into that executor.

 

How Executor Memory is Divided

It’s easy to think, “8 GB for the executor, I will get 8 GB for my data.” In fact, Spark splits that memory into some portions, and at the end of the day, each part plays its own role. Knowing these sections helps explain most memory surprises.

  • Spark itself reserves a small, fixed slice (about 300 MB); Spark’s memory must work. If after reserving this, there isn’t enough left, the executor won’t even get started. 
  • A good chunk, typically about 25 percent of the heap, is user memory. It includes our custom code: variables in user-defined functions, data structures you build inside transformations, accumulators, and metadata. Spark handles none of this space. But if you’re throwing large objects at it here, there might be a chance to blow up the memory without Spark having a chance to intervene.
  • The great bulk of heap is allocated, typically from 60-75%, to Spark’s unified memory pool that is monitored by the Unified Memory Manager. This is divided into:
    • Execution memory: temporary space for computations like joins, aggregations, sorts, and, most importantly, shuffles.
    • Storage memory: used to cache RDDs or DataFrames and store broadcast variables.

Spark uses a single shared memory pool for both execution and storage, and the two can dynamically borrow memory from each other when needed. By default, Spark aims for a 50/50 split (spark.memory.storageFraction = 0.5), but that split is not a hard wall. If storage is not using its full capacity, execution can take more, and if execution is quiet, storage can grow too.

The important part is what happens under pressure. Execution has higher priority, which means that if a task needs memory, Spark can evict cached blocks to free up space. Storage doesn’t have the same power; it cannot take memory away from running tasks. In practice, storage may expand during periods when execution is idle, but once computation begins, execution has the right to reclaim memory first.


What Happens when Execution Memory Becomes Limited?

When Spark runs a large join, sort, or shuffle, it needs temporary memory buffers to hold intermediate data during computation. When that memory starts running low, Spark doesn’t just fail immediately; it goes through a defined sequence to free up space and continue execution:

  1. Free some space: It drops cached data (things you .cache()/.persist() for reuse) from storage memory to free up more space for execution.
  2. And if that’s not good enough: Spark writes temporary data to disk (this is what is called spilling).
  3. Why you’re in pain: spilling prevents failures, but disk I/O is far slower than RAM. So big spills in a stage can quickly slow it down.

So, if you cache too much and run shuffle-heavy work, Spark may eventually have to evict caches and then spill which hurts performance. An easy way to reduce this pressure on memory is to use off-heap memory (moving certain buffers outside the JVM heap).

Heap vs Off-Heap Memory

Up to now, we’ve been talking about heap memory, the space inside the JVM that Spark manages. Heap memory is fast and easy, but the only disadvantage is that it is controlled by the Javagarbage collector. GC pauses can create severe slowdowns for a job with a long-running operation and large objects.

To handle this, Spark also has off-heap memory. Off-heap memory resides in memory outside the JVM and is managed directly by the operating system. Since it is not subject to garbage collection, it avoids GC overhead and lengthy stop-the-world delays. Spark uses it primarily for caching and shuffling data buffers. 

You enable it with two configs:

spark.memory.offHeap.enabled=true
spark.memory.offHeap.size=2g

Once enabled, Spark’s total usable memory = on-heap memory + off-heap memory. It helps Spark stay stable when shuffles or caching start to push memory.

Why Off-Heap Helps?

In Spark, two things usually eat the most memory. Caching and shuffling.

Caching
When calling cache() or persist() on a DataFrame, Spark keeps that data so it can reuse it later. If that cached data is in the JVM heap, Java’s garbage collector has more work. It needs to scan many objects. When the cached dataset is large, GC pauses get longer. They can also be random and difficult to predict. If Spark stores cached data off-heap instead, the heap remains cleaner.

Even when data is stored off-heap, Spark is still responsible for managing and clearing cached blocks. The OS provides raw memory, but Spark’s memory and storage managers decide when to evict or free it. GC does not need to deal with all those cached objects. This helps avoid “GC storms” when  caching large Tables.

Shuffling
During join or groupBy, Spark creates shuffle buffers in memory. These buffers are heavy. They also live for a short time. If they are on the heap, they create a lot of garbage quickly. This puts pressure on GC.
But if Spark keeps these buffers off-heap, it prevents the heap from being filled with temporary objects. That usually means less GC work. It also reduces the chance of heap OOM errors.

So in both cases, off-heap can make memory behavior more stable. It does not give you more RAM, but it reduces JVM overhead. And Spark becomes less sensitive to garbage collection. 

Limitations of off-heap memory

Off-heap is not a magic switch. It has trade-offs.

1) You must size it carefully

Off-heap memory is not taken from the executor heap, it is added on top of it. That means the executor will use more total memory. If you make it too large, you can run into memory limits even though the heap looks fine.

2) It does not help every workload
Off-heap is most useful for big caches and shuffle-heavy jobs. If your job is mostly simple transforms or CPU-heavy work, off-heap may not help, because you’re not spending much time on GC in the first place. In that case, it can even slow things down due to the extra serialization and memory-management work.

3) Executors still need a heap
Even with off-heap on, Spark still needs heap for many things. For example, task scheduling, broadcast variables, accumulators, and objects created by your code. You cannot shrink the heap too much.


Partitions: Controlling Task Size

Partitions decide how Spark splits data into tasks. One task usually works on one partition.

If you have too few partitions, each task becomes too big. One partition can hold gigabytes of data. That can overload an executor.
If you have too many partitions, Spark pays a different price. It must schedule thousands of tiny tasks. It also creates lots of small shuffle files. That adds overhead.

Spark uses defaults, but they are often not a good fit. File sources create partitions based on file splits. Shuffle operations, which we’ll explain in the next section, often default to 200 partitions. In real jobs, that number is rarely right.

Example: you are processing 1 TB of data. With 200 partitions, that is about 5 GB per partition. If an executor has 8 GB of memory, a task that uses 5 GB can push it over the limit.
If you use 2000 partitions instead, each partition is around 500 MB. That is much easier to handle.

So partitions are one of the main ways to control task size. If you pick them poorly, you either waste time on too many small tasks or you crash executors with tasks that are too large.


Shuffles: Where Memory Gets Tested

Partitions define how work is split across tasks. But during certain operations, Spark completely reshapes those partitions, a process called a shuffle. 

A shuffle happens when Spark needs to move data between executors so that rows with the same key end up on the same machine. This is typically triggered by operations like groupBy, join, or distinct.

For example,  say you have a dataset of purchases, and you want to count purchases by customer ID. All rows for a given customer must end up on the same executor so Spark can process them together. To make that happen, Spark reads data from many partitions, transfers it over the network, and then writes it again into new shuffle partitions.

If one key is much more common than the others, you get skew. One shuffle partition becomes huge. The executor that processes it can run out of memory.

Take this:

df.groupBy("country").count()

If the data is balanced, each country has roughly the same number of rows. Shuffle partitions stay similar in size.
But if 90% of rows are “US”, then the “US” partition gets almost everything. One reduce task gets overloaded. That executor can hit OOM. The rest of the cluster may sit and wait.

So that is why shuffles are points where a job fails.


Best Practices and Tuning Tips

Since we explained the Spark memory model, the next step is to keep memory use stable. That is what makes jobs fast and predictable.

1. Size executors for stability

Very large executors often suffer from long GC pauses. Very small executors run out of space. You can start with 4–5 cores and 16–32 GB heap per executor. Add 1–2 GB of overhead.
If you increase cores, increase memory as well.  All tasks on that executor share the same heap.

--conf spark.executor.cores=4
--conf spark.executor.memory=16g
--conf spark.executor.memoryOverhead=1536

2. Make partition sizes practical, not huge

Partitions decide how much data one task handles. Try to keep shuffle partitions around 100 MB to 1 GB each. Very large partitions can cause OOM. Very small ones add scheduling overhead. Start with the default 200 shuffle partitions. Increase only if tasks look too big.

3. Cache only when it pays off

Spark already keeps some working data in memory while it runs. Cache only if you reuse the same dataset. If you cached something and you’re done with it, call unpersist() early. If the reused data is very large, off-heap can help reduce GC pressure.

4. Control shuffles and handle skew

Shuffles regroup rows by key, and they stress memory and disk.

When a skew occurs, a single slow task can block the entire stage or even crash the job.  In some cases, a simple repartition()can already help distribute the data more evenly. If that’s not enough, there are more advanced techniques to handle skew.

Two common fixes are:

  • Salting (manual). Add a small, random suffix to the hot key (for example, customer_123_1 … _10). This spreads the rows across partitions. Then do the join or aggregation. After that, merge the results back to customer_123.
  • AQE (automatic). In Spark 3+, AQE watches partition sizes during the job. If one partition is huge, it can be split. If many are tiny, they can merge. 

5. Broadcast joins

Joins are causing huge shuffles. These shuffles move data across the cluster so matching keys land together, and that can use a lot of memory. It can also lead to OOM.

If one side is small, broadcasting is cheaper. Spark sends the small table to every executor. Then each executor joins it with its local data. No big shuffle. Much less network and memory pressure.

AQE can also switch a shuffle join to a broadcast join if a side table turns out to be small at runtime.  You can enable it and set a broadcast threshold:

--conf spark.sql.adaptive.enabled=true
--conf spark.sql.adaptive.skewJoin.enabled=true
--conf spark.sql.autoBroadcastJoinThreshold=134217728   # 128 MB


6. Use off-heap where it helps

Off-heap puts some buffers (shuffle, caching) outside the JVM heap. This usually reduces GC pauses. It does not replace the heap. Executors still need the heap for Spark internals and your objects.
Use it mainly for cache-heavy or shuffle-heavy workloads.

--conf spark.memory.offHeap.enabled=true
--conf spark.memory.offHeap.size=1g
--conf spark.executor.memoryOverhead=1536


7. Tune based on evidence (Spark UI)

Don’t tune blindly. Use the Spark UI after each run.

Executors tab

  • High GC time → executors are spending more time cleaning memory than doing work – usually too many objects in the heap.
  • “Java heap space” failures → not enough heap (or not enough overhead in some cases).
  • Uneven work across executors → often skew or uneven partitions.

SQL / Stages tab

  • A few tasks with huge input → skewed partitions.
  • High shuffle spills to disk → not enough execution memory, spilling will slow things down.
  • One or two tasks are much slower than the rest → skew, or partitions that are too large.

When you look at these signals, tuning becomes simpler. You can decide what to do next: more partitions, more memory, AQE, broadcast, or fixing skew.

 

Practical Example: Investigating Memory Issues in a Spark Job

Now let’s make this real. We’ll run a Spark job on a small local cluster. We’ll start it with Docker. The goal is to see memory pressure, not just talk about it.

Generating Example Data (Skew + Fat Rows)

We’ll create two JSON datasets that simulate a simple airline analytics scenario.

import org.apache.spark.sql.functions._

val OUT = "data/flights_json"
val ROWS = 10000000L
val PAD = 256
val SKEW = 0.9

// small dimension: airport → city
val airportDim = Seq(
("JFK","New York"), ("SFO","San Francisco"), ("LAX","Los Angeles"),
("SEA","Seattle"), ("DEN","Denver"), ("ORD","Chicago"),
("ATL","Atlanta"), ("DFW","Dallas"), ("PHX","Phoenix"),
("BOS","Boston"), ("IAD","Dulles"), ("MIA","Miami"), ("LAS","Las Vegas")
).toDF("airport","city")

val airports = airportDim.select("airport").as[String].collect()
val airlines = Array("UA","DL","AA","SW","BA","LH","AF","EK","QR","SQ")

val flights = spark.range(ROWS)
.withColumn("dep_airport", element_at(array(airports.map(lit):_*),
(rand()*airports.length).cast("int")+1))
.withColumn("arr_airport", when(rand(42) < SKEW, lit("JFK"))
.otherwise(element_at(array(airports.map(lit):),
(rand(99)*airports.length).cast("int")+1)))
.withColumn("airline", element_at(array(airlines.map(lit):_*),
(rand(7)*airlines.length).cast("int")+1)))
.withColumn("fare_usd", round(rand()*500 + 80, 2))
.withColumn("notes", rpad(lit("flight meta "), PAD, "x"))
.withColumn("flight_id", col("id") + 10000000)
.select("flight_id","dep_airport","arr_airport","airline","fare_usd","notes")

flights.write.mode("overwrite").json("data/flights_json")
airportDim.write.mode("overwrite").json("data/airport_dim")

What we created:

  • flights_json: ~100k flights, with skew (90% arrive to JFK → “New York”).
  • airport_dim: 13 airports with their cities (small dimension table).
  • The notes column is padded to make each row larger, stressing the shuffle buffers.

Here is data snippet:

{"flight_id":12500,"dep_airport":"DEN","arr_airport":"JFK","airline":"BA","fare_usd":323.18,"notes":"flight meta xxxxxxxxx...xxxxxxxxxx"}
{"flight_id":12501,"dep_airport":"ORD","arr_airport":"JFK","airline":"EK","fare_usd":513.09,"notes":"flight meta xxxxxxxxx...xxxxxxxxxx"}
{"flight_id":12502,"dep_airport":"BOS","arr_airport":"JFK","airline":"AA","fare_usd":459.15,"notes":"flight meta xxxxxxxxx...xxxxxxxxxx"}
{"flight_id":12503,"dep_airport":"PHX","arr_airport":"JFK","airline":"BA","fare_usd":489.71,"notes":"flight meta xxxxxxxxx...xxxxxxxxxx"}

{"airport":"ATL","city":"Atlanta"}
{"airport":"DFW","city":"Dallas"}
{"airport":"PHX","city":"Phoenix"}


Spark Job: Flights per City & Average Fare

Based on the above data, our goal is to find the number of flights and average fare per city.

val flights    = spark.read.json("/opt/spark-data/flights_json")
val airportDim = spark.read.json("/opt/spark-data/airport_dim")

val joined = flights.join(
airportDim,
flights("arr_airport") === airportDim("airport"),
"inner"
)

joined.cache()
joined.count()

val sorted = joined.orderBy(desc("fare_usd"))

val agg = sorted.groupBy($"city")
.agg(count("*").as("flights"), round(avg($"fare_usd"), 2).as("avg_fare"))
.orderBy(desc("flights"))

agg.collect()


This setup is meant to make problems easier to reproduce.

docker exec -it spark-master /opt/spark/bin/spark-submit \
 --class BadFlightsApp \
 --master spark://spark-master:7077 \
 --deploy-mode client \
 --total-executor-cores 4 \
 --executor-memory 1g \
 --driver-memory 2g \
 --conf spark.executor.memoryOverhead=384m \
 --conf spark.serializer=org.apache.spark.serializer.JavaSerializer \
 --conf spark.sql.adaptive.enabled=false \
 --conf spark.sql.autoBroadcastJoinThreshold=-1 \
 --conf spark.sql.shuffle.partitions=400 \
 /opt/spark-apps/spark-flights-bad-lab_2.12-0.1.0-SNAPSHOT.jar


Job Analysing

Running this job can result in slow execution, large shuffle reads and task time, high garbage collection, and, sometimes, even  OutOfMemoryError: Java heap space.

Now let’s explain why, using the Spark UI.

1. Shuffle Join on a Small Table

We disabled broadcast. So Spark treats the join like two big tables. It shuffles both sides.

In Spark UI:

  • SQL tab shows SortMergeJoin (physical join algorithms used when joining two large datasets, which requires both sides to be sorted and shuffled)
  • large Shuffle Read (186 MiB).

Impact:

  • Increased I/O, more memory buffers needed, and a higher risk of spills.


2. Global Sort Creates Another Shuffle

This line is expensive:

orderBy(desc("fare_usd"))

A global sort forces a full shuffle. Spark must move rows so partitions get the correct order range. And in this job, we don’t need it. We only want per-city aggregation.

Impact:

  • More shuffle, more memory.

3. Storing Deserialized Objects on the Heap

cache() defaults to MEMORY_ONLY. That means “deserialized Java objects in the heap”. With wide rows (like our padded notes), this grows fast. GC then has many objects to scan.

In Spark UI:

  • Storage tab shows Memory Deserialized.

Impact:

  • High GC overhead and unpredictable pauses. Cached data may also be evicted when Spark needs execution memory.


4. Skew on “JFK”

About 90% of rows have arr_airport = JFK. That becomes city = New York. During groupBy, those rows land in the same reducer partition. One task becomes huge.

In Spark UI:

  • Stage details: one task runs far longer than others.
  • Task metrics show one partition with a huge Shuffle Read and high spill (disk).

Impact:

  • One executor is overloaded while others are idle.
  • A common cause of OutOfMemoryError or a slow final stage.


The Optimized Version

Here’s the same logic, but with better parameters and Spark features:

val flights    = spark.read.json("/opt/spark-data/flights_json")
val airportDim = spark.read.json("/opt/spark-data/airport_dim")

// broadcast the small dimension → removes one shuffle
val joined = flights.join(
broadcast(airportDim),
flights("arr_airport") === airportDim("airport"),
"inner"
)

// serialized cache → less GC
val projected = joined.persist(StorageLevel.MEMORY_ONLY_SER)
projected.count()

// This still causes a shuffle because repartition($"city") redistributes the data across partitions. However, it avoids the more expensive global sort from orderBy(...) by sorting only within each partition.
val localSorted = projected.repartition($"city").sortWithinPartitions(desc("fare_usd"))

// enable Adaptive Query Execution (AQE)
spark.conf.set("spark.sql.adaptive.enabled", true)
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", true)

// same aggregation
val agg = localSorted.groupBy($"city")
.agg(count("*").as("flights"), round(avg($"fare_usd"), 2).as("avg_fare"))
.orderBy(desc("flights"))

agg.collect()


Run with more memory and better serializer:

docker exec -it spark-master /opt/spark/bin/spark-submit \
--class GoodFlightsApp \
--master spark://spark-master:7077 \
--deploy-mode client \
--total-executor-cores 4 \
--executor-memory 1g \
--driver-memory 2g \
--conf spark.executor.memoryOverhead=384m \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
--conf spark.sql.adaptive.enabled=true \
--conf spark.sql.adaptive.skewJoin.enabled=true \
—-conf spark.sql.shuffle.partitions=400 \
/opt/spark-apps/spark-flights-good-lab_2.12-0.1.0-SNAPSHOT.jar

What we changed (and why it helps)

  1. Broadcast the tiny table
    The small airport_dim is sent to all executors, so the large flights table is not shuffled. This reduces network traffic and avoids sort-merge joins.
  2. Enable AQE with skew handling
    Adaptive Query Execution splits the oversized “JFK” group and merges tiny partitions. Tasks finish in similar times, with no end-of-job straggler.
  3. Remove the global sort before aggregation
    Dropping the early orderBy prevents an extra, full-dataset shuffle. A small sort after the aggregation is inexpensive.
  4. Use serialized caching
    Caching in MEMORY_ONLY_SER stores rows compactly and reduces GC pressure. In our run, the cache fit in memory and did not spill to disk.

And this is how it looks in SparkUI:



Results and Summary

After running both configurations, we can see how tuning Spark’s memory and shuffle behavior affects performance. The table below summarizes the main metrics and the reasons behind each improvement.


Final Thoughts

Spark memory errors are not random. They usually mean something is out of balance. Most of the time, the problem is between executors, shuffles, and partitions.

Executors have limited memory. Spark splits it into a few parts. Some are reserved. Some is for your code. Some are for Spark itself. Shuffles can burn through that memory fast. If it does not fit, Spark spills to disk. Partitions decide how big each task is. If partitions are uneven, one task can take too much time and fail.

Once you understand this, Spark feels less “mysterious”. You stop guessing. You don’t just add more RAM and hope. You can change specific settings, such as executor size, number of partitions, caching, and AQE.

The main idea is balance. Executors, shuffles, and partitions work together. If one is off, the job slows or becomes unstable.  If they match well, Spark runs smoothly.

Also, tuning is not something you do once. You run the job, check the Spark UI, and adjust as needed. Then you test again. Over time, the job becomes stable and fast, even on large data.

 

Leave a comment

Your email address will not be published. Required fields are marked *