What is spill in Spark?

Summary

• Spill is what happens when a Spark task runs low on execution memory, the memory used for shuffles, joins, sorts and aggregations, and moves part of its in-memory data to local disk.
• The Spark docs say spilling incurs "the additional overhead of disk I/O and increased garbage collection", so a stage that spills takes longer.
• The Spark UI reports it as "Shuffle spill (memory)" and "Shuffle spill (disk)" on the stage page. The tuning guide's simplest fix for oversized task working sets is more parallelism, so each task's input set is smaller.

What is spill in Spark?

Spill is what happens when a Spark task runs low on execution memory and writes part of its working data to local disk. The Spark tuning guide defines execution memory as the memory "used for computation in shuffles, joins, sorts and aggregations". Those operations build in-memory data structures, such as hash tables for grouping. When the data doesn't fit, the RDD programming guide says Spark "will spill these tables to disk, incurring the additional overhead of disk I/O and increased garbage collection."
So spill is a cost you pay in time. Most documented fixes shrink what each task has to hold in memory.

Which memory runs out when Spark spills?

Execution and storage (cached data) share one unified region of the executor heap. spark.memory.fraction sets its size as a fraction of (JVM heap space - 300MiB), default 0.6. "Execution may evict storage if necessary", but only down to a protected share set by spark.memory.storageFraction (default 0.5), and "Storage may not evict execution".
The configuration reference ties both settings to spill. Lowering spark.memory.fraction means "the more frequently spills and cached data eviction occur". Raising spark.memory.storageFraction means "tasks may spill to disk more often." For both, "Leaving this at the default value is recommended."

Which operations spill?

• Shuffle writes. "Results from individual map tasks are kept in memory until they can't fit." Then they are sorted by target partition and written to a file.
• Key-based aggregations. reduceByKey and aggregateByKey build their structures on the map side, and 'ByKey operations build them on the reduce side.
• Joins and grouping. Shuffle operations such as sortByKey, groupByKey, reduceByKey and join "build a hash table within each task to perform the grouping, which can often be large."
• Spark SQL sorts and hash aggregations. The Sort and HashAggregate operators report their own spill.

How do you see spill in the Spark UI?

Metric	What it is	Where to find it
Shuffle spill (memory)	"the size of the deserialized form of the shuffled data in memory"	Stage detail page
Shuffle spill (disk)	"the size of the serialized form of the data on disk"	Stage detail page
Peak execution memory	"the maximum memory used by the internal data structures created during shuffles, aggregations and joins"	Stage detail page
spill size	"number of bytes spilled to disk from memory in the operator"	SQL tab, Sort and HashAggregate nodes
memoryBytesSpilled, diskBytesSpilled	In-memory and on-disk bytes "spilled by this task"	REST API, /applications/[app-id]/stages

The stage page also breaks these metrics down by executor and per task, so you can see whether a few tasks spill or all of them do.

How do you reduce spill?

Option	When to use it	Trade-off
Raise parallelism	Shuffle partitions are large. For DataFrames, set spark.sql.shuffle.partitions (default 200).	More tasks, but Spark "has a low task launching cost"
Keep AQE on	spark.sql.adaptive.enabled and spark.sql.adaptive.coalescePartitions.enabled (both default true) merge small post-shuffle partitions.	Start from "a large enough initial number of shuffle partitions"
Split skewed joins	A few tasks hold most of the data. spark.sql.adaptive.skewJoin.enabled (default true) splits them "into roughly evenly sized tasks".	Shuffled joins only (sort-merge and shuffled hash)
Broadcast the small side	One join input is small. spark.sql.autoBroadcastJoinThreshold (default 10MB) sets the automatic limit.	The table goes to every worker node
Add executor memory	Partitions are already small. Set spark.executor.memory (default 1g).	Large heaps may need a bigger G1 region size
Cache only what you reuse	Apps without caching "can use the entire space for execution, obviating unnecessary disk spills"	You recompute what you don't cache

from pyspark.sql import functions as sf

# More shuffle partitions means less data per task
spark.conf.set("spark.sql.shuffle.partitions", "400")

# Mark the small side for a broadcast join
joined = orders.join(sf.broadcast(countries), "country_code")

Don't reach for coalesce() here. It skips the shuffle, but going from 1000 to 100 partitions means "each of the 100 new partitions will claim 10 of the current partitions", so each task holds more data.

How Databricks approaches spill

The Databricks Spark UI guide checks spill first in a long-running stage, before skew and I/O. It calls spill "what happens when Spark runs low on execution memory" and notes "It is most common during data shuffling." Databricks also offers auto-optimized shuffle: setting spark.sql.shuffle.partitions to auto "automatically determines this number based on the query plan and the query input data size". Serverless compute "does not support setting most Spark properties", but this one is settable and defaults to auto.

FAQs

What's the difference between Shuffle spill (memory) and Shuffle spill (disk)?

They are two sizes of the same spill: the deserialized size of the data in memory and its serialized size on disk. Neither measures cached data, which the Storage tab lists.

Should I raise spark.memory.fraction to stop spill?

Usually not. The docs recommend the default, and the rest of the heap holds user data structures, internal metadata and a safeguard against out-of-memory errors from unusually large records. Add partitions first.

Is spill the same as an OutOfMemoryError?

No. Spill means Spark moved data to disk. An OutOfMemoryError can still happen when "the working set of one of your tasks, such as one of the reduce tasks in groupByKey, was too large." The tuning guide's simplest fix is the same: "increase the level of parallelism, so that each task's input set is smaller."