Why is my job spilling to disk?

Summary

• A Spark job spills when an operator that uses execution memory (a sort, a hash aggregation, a shuffle write, a sort-merge join) can't fit its partition in the memory available to its task, so it writes the overflow to local disk and merges it back later.
• The usual causes are partitions that are too large, skewed keys that overload a few tasks, and less execution memory per task than you assumed, often because of cached data or many tasks sharing one executor.
• Confirm it in the Spark UI: the Stage detail page reports Shuffle spill (memory) and Shuffle spill (disk) per task, and the SQL tab shows a "spill size" metric on Sort and HashAggregate operators.
• The first fix is usually more, smaller partitions. Spark's tuning guide calls increasing the level of parallelism "the simplest fix" when a task's working set is too large; raise spark.sql.shuffle.partitions (default 200) or let AQE start from a large initial partition count.
• Leave spark.memory.fraction at its default of 0.6, as the Spark docs recommend. On Databricks, setting spark.sql.shuffle.partitions to auto turns on auto-optimized shuffle.

Why is my job spilling to disk?

Because an operator ran out of execution memory for the partition it was processing. Sorts, hash aggregations, shuffle writes and sort-merge joins keep their working data in memory, and when a task's share isn't enough, Spark writes the overflow to local disk and merges it back later. That's a fallback rather than a failure: it lets the operator finish when memory runs short. Each spill still adds serialization and disk I/O, so stages with heavy spill run slowly. In most jobs the cause is one of three things: partitions that are too big, skewed keys, or less execution memory per task than you assumed.

How Spark decides when to spill

Spark gives execution and storage one shared region, M, sized by spark.memory.fraction (default 0.6) as a fraction of the JVM heap minus a 300 MiB reserve. Execution covers shuffles, joins, sorts and aggregations. Storage covers cached data. Execution can evict cached blocks, but only down to a protected subregion, R, sized by spark.memory.storageFraction (default 0.5), and storage can never evict execution. The configuration docs spell out the consequence of shrinking M: "The lower this is, the more frequently spills and cached data eviction occur."

How to confirm where it spills

• Stage detail page. Summary metrics include Shuffle spill (memory), "the size of the deserialized form of the shuffled data in memory," and Shuffle spill (disk), "the size of the serialized form of the data on disk." Compare the median task with the largest one. Spill on nearly every task points to partition size; spill on a handful points to skew.
• SQL tab. The "spill size" metric on Sort and HashAggregate nodes tells you which operator spilled.
• REST API. The monitoring API exposes memoryBytesSpilled and diskBytesSpilled, which helps for jobs you can't watch live. From a classic PySpark session:

import requests

ui, app = spark.sparkContext.uiWebUrl, spark.sparkContext.applicationId
for s in requests.get(f"{ui}/api/v1/applications/{app}/stages").json():
    if s["diskBytesSpilled"] > 0:
        print(s["stageId"], s["name"], s["memoryBytesSpilled"], s["diskBytesSpilled"])

Causes and fixes

Symptom	Likely cause	Fix
Nearly every task in a shuffle stage spills	Shuffle partitions too large for each task's memory	Raise spark.sql.shuffle.partitions (default 200), or set a large spark.sql.adaptive.coalescePartitions.initialPartitionNum and let AQE coalesce down
A stage that reads files spills	Input partitions too large	Lower spark.sql.files.maxPartitionBytes (default 128 MB)
A few tasks spill heavily, the rest are clean	Skewed keys	Keep spark.sql.adaptive.skewJoin.enabled on (default true) for sort-merge joins; pre-aggregate or split hot keys in aggregations
"spill size" on a Sort node feeding a join	Sort-merge join over a large input	Broadcast the small side; spark.sql.autoBroadcastJoinThreshold defaults to 10 MB
Spill started after you cached large DataFrames	Cached blocks inside R can't be evicted by execution	unpersist() anything you no longer reuse
Everything spills on small executors	Little execution memory per task	Raise spark.executor.memory or lower spark.executor.cores, because tasks running together on an executor share its memory
Local disks fill up during the job	Scratch space on a small volume	Point spark.local.dir (or SPARK_LOCAL_DIRS / LOCAL_DIRS) at larger, fast local disks

The tuning guide's advice for oversized task working sets is direct: "The simplest fix here is to increase the level of parallelism, so that each task's input set is smaller." In code, the three most common fixes look like this:

from pyspark.sql import functions as F

spark.conf.set("spark.sql.shuffle.partitions", "800")        # more, smaller partitions
joined = orders.join(F.broadcast(customers), "customer_id")  # no sort of the big side
lookup.unpersist()                                           # free cached data

Then recheck the Stage detail page. If spill moved from every task to a few, the remaining problem is skew.

How Databricks approaches spill

The diagnosis is the same on Databricks. One difference: setting spark.sql.shuffle.partitions to auto enables auto-optimized shuffle, which "automatically determines this number based on the query plan and the query input data size." Serverless notebooks and jobs default to auto and allow only a short list of Spark properties, including spark.sql.shuffle.partitions and spark.sql.files.maxPartitionBytes, so the partition-size fixes still apply while executor memory and cores aren't yours to set.

FAQs

Is spilling to disk a problem?

Not by itself. It's how an operator finishes when its data doesn't fit in memory, and it only needs fixing when it dominates a stage's run time.

Should I raise spark.memory.fraction?

Rarely. The docs recommend leaving it at the default, because the rest of the heap holds user data structures and internal metadata and absorbs imprecise size estimates for unusually large records. Fix partition size and skew first.

Doesn't AQE handle this automatically?

Partly. AQE has been on by default since Spark 3.2.0. It coalesces small shuffle partitions, and it splits a skewed sort-merge join partition when it's larger than both spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5.0) times the median and spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256MB). Its skew handling targets joins, so a hot key in an aggregation still needs your attention.