Spark spill to disk: when it happens and how to stop it

Spark spills to disk when one task's sort, join or aggregation no longer fits in the execution memory it may use. That memory is (executor memory − 300 MiB) × 0.6, and the tasks running on the executor share it: with 4 running, each may take at most a quarter. The fix is almost always smaller partitions, not more memory.

The rest of this article measures it on a real job, shows where each spill number comes from, and gives the fixes in the order to try them.

The memory a task may use

An executor's Java heap is split by spark.memory.fraction, 0.6:

1. 300 MiB reserved for Spark itself.
2. 60% of the rest, the unified pool, shared by execution (sorts, joins, aggregations, shuffle buffers) and cached data. Execution can push cached data out; with nothing cached, it has the whole pool.
3. The other 40% for your own objects and Spark's metadata.

Then the share: with N tasks running on the executor, each may take at least 1/2N and at most 1/N of the pool before it has to spill. Tasks run 1 per core, so N is usually spark.executor.cores.

For an 8 GiB executor with 4 cores: (8,192 − 300) × 0.6 = 4,735 MiB, so each task may hold about 1,184 MiB. More on the split in Spark executor memorySpark Executor MemoryThe memory each Spark worker process gets: a Java heap, split into a shared pool for shuffles, joins and cached data and a pool for your own objects, plus overhead outside the heap..

A job that spills

We ran a sort of 12 million orders on Spark 4.2.0, in local mode with 1 GiB of driver memory and 4 cores. Unified memory was (1,024 − 300) × 0.6 = 434 MiB, so each of the 4 tasks could hold about 108 MiB. With spark.sql.shuffle.partitions set to 4, the executor log showed each task hitting that limit again and again:

INFO UnsafeExternalSorter: Thread 86 spilling sort data of 108.0 MiB to disk (0 time so far)
INFO UnsafeExternalSorter: Thread 86 spilling sort data of 108.0 MiB to disk (1 time so far)
INFO UnsafeExternalSorter: Thread 86 spilling sort data of 108.0 MiB to disk (2 times so far)

108.0 MiB is exactly a quarter of the pool. Then we set 64 partitions and ran it again:

Same data, same memory, no spill. Each task read about 15 times less, and every task's sort fitted in the 108 MiB it was allowed.

Notice the gap between columns 1 and 3: each task read 51.7 MiB of shuffle data but needed about 336 MiB to sort it, 108 MiB held plus 228 MiB spilled. Shuffle bytes are compressed; rows in memory are not. That is why dividing the shuffle size by the memory per task underestimates the partitions you need.

Spill (memory) and Spill (disk)

The Spark UI shows 2 spill figures for each stage and task. They describe the same spilled data:

• Spill (memory) is its size as rows in memory, when Spark decided to write it out.
• Spill (disk) is its size once serialized and compressed on disk (spark.shuffle.spill.compress is on by default).

So Spill (disk) is always the smaller of the 2: 72.8 MiB against 228 MiB in our run. Neither figure is an amount of memory you are short of. Spill (memory) adds up every piece a task wrote, over every time it spilled.

Find where it spills

• On the stage page, the summary metrics give Spill (memory) and Spill (disk) from min to max. Similar on every task means partitions are too big for the memory. 0 at the median and a lot at the max means 1 heavy key: data skewShuffle and Data SkewA shuffle moves rows between machines so rows with the same key meet, for a join or group by; skew is when one key has far more rows, so one task does most of the work..
• In the SQL / DataFrame tab, the Sort and HashAggregate boxes show spill size and peak memory, so you see which step of the query spilled. A sort spills for a sort-merge join or an orderBy, an aggregate for a groupBy.
• In the executor log, look for spilling sort data of ... to disk, with the count so far.

Our Spark UI guide opens this very job, before and after the fix, and points at each of these numbers. Try your own sizes here:

Try itWill this stage spill to disk?

Executor memory (GiB)

8

1832

Cores per executor

4

48

Shuffle read of the stage (GiB)

200

102001,000

Shuffle partitions

200

2002,000

Rows in memory, against shuffle bytes

2 times4 times6 times

What 1 task holds4.0 GiB

What 1 task may use1.2 GiB

The fixes, in order

1. More shuffle partitions. The most direct fix, as Spark's tuning guide says: more parallelism, less data per task. Aim for partitions whose rows fit in the memory per task, and remember rows in memory are several times the shuffle bytes. Set it for the job, before the query that spills:

   spark.conf.set("spark.sql.shuffle.partitions", 64)
   

   Our guide to choosing shuffle partitions works out the number, and the shuffle partition calculator does the sums.

2. Split the heavy key. If only 1 task spills, more partitions do not help: every row of the key still lands in 1. In a join, adaptive query execution splits a partition over 5 times the median and over 256 MB; check that spark.sql.adaptive.skewJoin.enabled is on. In a groupBy it cannot, so add a salt to the key or deal with the heavy value apart. How to handle skewed data in PySpark shows both.

3. Carry fewer bytes. Select only the columns the join or sort needs, filter before the shuffle, and leave wide strings and arrays out until after it.

4. More memory per task. Fewer cores per executor give each running task a bigger share; more executor memory grows the pool. Both cost money, so do them last. The Spark sizing calculator picks executor sizes for a workload.

5. Leave spark.memory.fraction alone. Raising it gives execution more of the heap, but Spark's docs recommend the default: the rest holds metadata, your objects and room for size estimates that are off. Too high, and you trade spill for out-of-memory errors.

When spill turns into failure, Spark out of memory errors matches each error to its fix.

The settings that matter

There is no setting to turn spill off. Spill is what lets a task finish when its data does not fit; the settings above decide how often it has to.