Spark Executor Memory: Spill, GC, and the OOM Errors That Are Really Partition-Sizing Errors

ExecutorLostFailure: Container killed by YARN for exceeding memory limits. 30.2 GB of 30 GB physical memory used. I have read that line more times than any other in this job, and it has almost never meant what the team thought it meant.

The default reaction is to raise spark.executor.memory. Sometimes that works for a week. Then the data grows 15 percent, the same error comes back, and now you are paying for 60 GB executors to process 400 GB of data. I have walked into environments running r5.12xlarge nodes for jobs that should fit comfortably on r5.2xlarge, entirely because every OOM was answered with more RAM.

The actual cause is usually that one partition is too big for one task, or that off-heap usage grew outside the JVM where executor.memory has no effect. Those are different failures with different fixes, and the error message tells you which one you have if you know how the memory is divided.

So let us divide it, then work through the six failure modes I can now identify from the stack trace before opening the UI.

How executor memory is actually split

A Spark executor container has four regions, and people conflate them constantly.

Reserved memory — 300 MB, hardcoded, for Spark’s own bookkeeping. You cannot touch it.

Unified memory — (executor.memory - 300MB) * spark.memory.fraction, default fraction 0.6. This is the interesting part, and it is itself split into two halves that borrow from each other:

• Execution memory holds shuffle sort buffers, hash tables for joins and aggregations, and sorting workspace. When it runs out, Spark spills to disk.
• Storage memory holds cached RDDs and DataFrames plus broadcast variables. When it runs out, cached blocks are evicted.

The borrowing is asymmetric and this matters: execution can evict storage, but storage cannot evict execution. spark.memory.storageFraction (default 0.5) sets the storage share that execution may not steal.

User memory — the remaining 40 percent. Your UDF objects, whatever you build in a mapPartitions, and any data structure created outside Spark’s managed structures.

Overhead memory — spark.executor.memoryOverhead, default the larger of 384 MB or 10 percent of executor memory. This lives outside the JVM heap: PySpark worker processes, off-heap Arrow buffers, netty network buffers, and native libraries. This is the region that gets containers killed.

flowchart TD
  cont["Executor container"] --> jvm["JVM heap: executor.memory"]
  cont --> ovh["Overhead: off-heap, python, netty"]
  jvm --> res["Reserved 300 MB"]
  jvm --> uni["Unified pool: fraction 0.6"]
  jvm --> usr["User memory: 0.4"]
  uni --> exe["Execution: sort, join, agg"]
  uni --> sto["Storage: cache, broadcast"]
  exe -.->|"can evict"| sto
  ovh --> pyw["PySpark workers live here"]

Raising executor.memory grows the JVM heap. It does nothing for the overhead region where PySpark and Arrow actually live.

For a PySpark job, the Python interpreters running your UDFs are in the overhead region. That single fact explains most of the confusing OOMs in Python-heavy pipelines.

The number that matters: bytes per task

A Spark task processes one partition, in one core, in whatever execution memory that core’s share allows. With a 16 GB executor and 4 cores, unified memory is about 9.4 GB, execution memory is roughly half of that under contention, so each task has on the order of 1.2 GB of execution workspace.

Feed that task a 6 GB partition and it will spill, repeatedly. Feed it a 6 GB partition that must also be hash-aggregated into a large map, and it will OOM.

So the first question on any OOM is not “how much memory does the executor have” but “how big is the largest partition.” Measure it:

from pyspark.sql import functions as F

df = spark.read.table("prod.silver.events")

# Bytes per partition, estimated from the plan
print(df._jdf.queryExecution().optimizedPlan().stats().sizeInBytes()
      / df.rdd.getNumPartitions() / 1024**2, "MB per partition (avg)")

# Actual row distribution across partitions: the honest version
counts = (df.withColumn("pid", F.spark_partition_id())
            .groupBy("pid").count()
            .orderBy(F.desc("count")))
counts.show(10)

Average partition size lies when data is skewed. The partition-level count is what I trust, and if the max is many times the median, the OOM is a skew problem and belongs in the skew playbook rather than a memory ticket.

My target is 128-200 MB of data per task at every stage boundary. At that size, spill is occasional and OOM is rare with default memory settings.

Reading spill correctly

Spill is Spark working as designed: execution memory filled, so sorted runs went to local disk to be merged later. It is not an error. It is, however, expensive, and the UI gives you two numbers.

Shuffle Spill (Memory) is the in-memory size of data that got spilled — deserialized, so it looks large. Shuffle Spill (Disk) is the serialized size actually written. The ratio between them is just serialization efficiency, and the number I act on is disk spill compared to shuffle write.

• Disk spill near zero: partitions fit. Nothing to do.
• Disk spill up to about 30 percent of shuffle write: acceptable on big aggregations, mild slowdown.
• Disk spill exceeding shuffle write: every byte is now written at least twice and read twice. This is where 40-minute stages come from, and where I intervene.

The intervention is almost always more partitions, not more memory:

# AQE handles this at runtime for shuffle stages
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

# For the initial read stage, AQE cannot help. Control input splits.
spark.conf.set("spark.sql.files.maxPartitionBytes", "128MB")

# Fewer cores per executor means more execution memory per task,
# which is the right trade for memory-hungry aggregations
# 4 cores x 16 GB beats 8 cores x 32 GB for spill-heavy work

That last point is counterintuitive and worth internalizing. Cores per executor divide the same unified pool. An 8-core executor with 32 GB gives each task the same workspace as a 4-core executor with 16 GB, but with double the contention and worse GC pauses. I default to 4-5 cores per executor and have never regretted it.

Six OOM failures and what each one means

1. java.lang.OutOfMemoryError: Java heap space in an executor. The JVM heap filled. Cause is nearly always a partition too large for the task, or a groupByKey-style operation materializing all values for one key. Fix: more partitions, or rewrite the aggregation to be partial-then-final.

2. Container killed by YARN for exceeding memory limits. The container exceeded its total limit, including overhead. The JVM was fine. Fix: raise spark.executor.memoryOverhead. For PySpark jobs with pandas UDFs I run 25-30 percent of executor memory as overhead rather than the 10 percent default.

3. OutOfMemoryError: GC overhead limit exceeded. The JVM spent 98 percent of time in GC and reclaimed almost nothing. Usually means live objects nearly fill the heap. Same fix as #1, plus check for a large collection accumulating in a UDF closure.

4. Driver OOM. Someone called collect(), toPandas(), or broadcast a table that was not small. This is a code review finding.

# Every one of these pulls the full result to the driver
rows = df.collect()                    # do not
pdf  = df.toPandas()                   # do not, unless tiny
big  = spark.table("prod.silver.events")
joined = fact.join(F.broadcast(big), "id")   # 3 TB broadcast, driver dies

# Safe versions
sample = df.limit(1000).toPandas()     # bounded
df.write.mode("overwrite").saveAsTable("scratch.result")  # let executors write

5. MemoryError inside a Python worker. A pandas UDF received a batch too large to materialize as a pandas DataFrame. Lower spark.sql.execution.arrow.maxRecordsPerBatch from its 10,000 default, and raise overhead memory.

# Arrow batch size is the knob for pandas UDF memory, not executor.memory
spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", "2000")
spark.conf.set("spark.executor.memoryOverhead", "4g")

6. SparkOutOfMemoryError: Unable to acquire N bytes of memory. Execution memory could not be acquired even after spilling — typically a single record or key group that is genuinely too large, like a 2 GB JSON blob in one row. Fix the data, not the config.

Sizing an executor from scratch

When I size a new cluster I work from the node, not from a guess. For a 64 GB, 16-core node:

• Leave 1 core and about 8 percent of memory for the OS and node agents.
• Choose 4 cores per executor, giving 3 executors per node.
• Divide the remaining memory by 3, then reserve overhead: roughly 16 GB per executor with 4 GB of that as overhead for PySpark workloads, or 1.5 GB for pure Scala.

# Scala-heavy job
spark.conf.set("spark.executor.cores", "4")
spark.conf.set("spark.executor.memory", "16g")
spark.conf.set("spark.executor.memoryOverhead", "2g")

# Same node, PySpark with pandas UDFs: shift memory to overhead
spark.conf.set("spark.executor.cores", "4")
spark.conf.set("spark.executor.memory", "12g")
spark.conf.set("spark.executor.memoryOverhead", "6g")

The PySpark version looks wrong to people used to JVM tuning. It is not: the Python processes need real memory and they do not live in the heap. Under-provisioning overhead is the number one cause of mysterious container kills in Python pipelines, and it is why I ask “is this job Python or Scala” before any memory conversation.

On Databricks, most of this is set for you by instance type, and the job cost conversation becomes the more useful one. But the partition-sizing logic is identical regardless of who picks the instance.

Pitfalls

Answering every OOM with more memory. It works once, then the data grows. If the underlying issue is a 6 GB partition, you are buying time at a linear cost with no end.

Raising executor.memory to fix a container kill. Container limits include overhead, so growing the heap inside a fixed container makes the kill more likely, not less. Raise overhead.

Eight or more cores per executor. More tasks share the same unified pool, GC pauses get longer, and spill increases. Four to five is the range I have found stable across every cluster type.

Caching aggressively. Cache consumes storage memory that execution would have used, and execution cannot evict the protected storageFraction. Cache only DataFrames read more than twice, and unpersist them explicitly.

Ignoring spill because the job succeeds. Disk spill exceeding shuffle write is a 2-3x slowdown hiding in a green pipeline. It is the cheapest performance win most teams have available.

Treating maxPartitionBytes as irrelevant because AQE is on. AQE only reshapes shuffle stages. The initial file scan partition size is set by spark.sql.files.maxPartitionBytes, and a table of 1 GB Parquet files will hand you 1 GB tasks.

FAQ

Should I change spark.memory.fraction?

Rarely. The 0.6 default is well chosen. If a job uses heavy custom objects in mapPartitions, lowering it to 0.5 gives user memory more room; if the job is pure SQL with no UDFs, 0.7 can reduce spill. Both are small effects compared to partition sizing.

Is off-heap memory worth enabling?

For very large shuffles and Tungsten-heavy SQL, spark.memory.offHeap.enabled with a sized offHeap.size reduces GC pressure measurably. It also adds a second memory pool to reason about, so I enable it only after partition sizing is right and GC time is still above 10 percent.

Why does the same job OOM on a bigger cluster?

Because parallelism changed the partition count. Fewer, larger partitions per task, or more concurrent tasks per executor sharing the pool. Cluster size and partition size are independent knobs, and only the second one determines what a task must hold.

How much overhead memory for PySpark?

Start at 25 percent of executor memory, more if you use pandas UDFs or large Python dependencies. The Python worker count equals the executor core count, so the per-worker footprint multiplies.

Does the Delta or Iceberg format affect memory?

Indirectly, through file sizes. Very large data files produce large scan partitions unless maxPartitionBytes splits them, and the row-group layout affects how much a task buffers. Table maintenance that keeps files near 128-256 MB, as described in the transaction log post, helps memory as much as it helps read pruning.

What this means for your pipelines

Stop treating OOM as a capacity signal. Read the exact error text first: heap versus container kill versus Python MemoryError versus driver, because each points at a different region and only one of them is fixed by spark.executor.memory. That triage takes 30 seconds and saves a week of cluster inflation.

Then make partition size an explicit, monitored property of your jobs. Set maxPartitionBytes for the scan stage, let AQE hold shuffle stages near 128 MB, keep 4-5 cores per executor, and provision overhead honestly for Python workloads. With those four things in place, default memory settings handle a surprising range of data volumes without touching a single memory config.

Finally, watch spill as a routine metric alongside runtime. It is the earliest signal that data growth has pushed partitions past what tasks can hold, and it shows up weeks before the first OOM page. If you are already tracking shuffle bytes from the shuffle post, adding spill to the same dashboard is an afternoon of work and the best early warning system you will build this quarter.