Spark Data Spill: Why 200 Partitions Made My Job Take 49 Minutes

I ran a global ORDER BY on 8 billion records.

I expected it to be expensive. A global sort has to move data across the cluster before Spark can guarantee that the final output is ordered.

But I did not expect one line of code to generate approximately 332 GB of shuffle data and keep the job running for 49 minutes.

Nothing appeared broken. There were no failed tasks, no obvious skew, and Adaptive Query Execution was already enabled. Even the cluster looked healthy.

That was the interesting part.

The query was slow, but Spark was not showing the kind of failure that normally points directly to the problem. So instead of changing the cluster, I opened the Spark UI and followed the data movement.

Three numbers changed the direction of the investigation:

• Input reported across the sort workflow: approximately 263 GB
• Shuffle write: approximately 332 GB
• Initial shuffle partitions: 200

Once I put those numbers together, the bottleneck became much easier to see.

332 GB / 200 partitions = approximately 1.66 GB per shuffle partition

Each task was being asked to sort approximately 1.66 GB of shuffle data on average. On this cluster, that was more than a concurrent task that could comfortably hold in execution memory.

Spark did not fail. It did exactly what it was designed to do when the intermediate data could not remain in memory.

It quietly started spilling to disk.

I increased spark.sql.shuffle.partitions from 200 to 2,700. The observed average partition size fell to around 140 MB, spill reduced sharply, and the same workload completed in 27 minutes.

The configuration change was simple. Finding the reason it worked required following the complete story inside the Spark UI.

I Expected ORDER BY to Be Expensive. Not This Expensive.

The experiment used a Delta table containing approximately 8 billion records and occupying about 131 GB in cloud storage.

I ran it on four Standard_D4ds_v5 worker nodes. Together, the workers provided 16 CPU cores and 64 GB of physical memory, with four cores and 16 GB of RAM on each worker.

The query itself was deliberately simple:

I used the noop format so Spark would execute the complete sort without adding the cost of writing a normal Delta or Parquet output. That kept the experiment focused on the operation I wanted to study.

At first glance, the code does not look capable of creating a 49-minute job. There is no join, aggregation, window function, or explode.

But ORDER BY is a wide transformation. Spark cannot sort the records inside the existing partitions and stop there. It must also make sure that the partitions themselves represent the correct global range.

To do that, Spark estimates boundaries for the sort key, redistributes the rows using those boundaries, and sorts the records inside each resulting partition. Creating the range boundaries can also require sampling before the main shuffle begins.

So those few lines of code could trigger sampling, another input pass, a full range-based shuffle, and a sort inside every final partition.

That explained why the job involved data movement. It still did not explain why it was taking 49 minutes.

The Spark UI Showed More Work Than the Code Suggested

The first clue appeared in the input metrics.

The Delta table occupied approximately 131 GB in cloud storage, but the complete SQL workflow reported close to 263 GB of input across its jobs and stages.

That did not mean Spark had duplicated the table. In this run, the additional input work appeared around the sampling and main processing required for the global range sort.

The exact accounting can vary with the physical plan, runtime, caching, and data source, so I treat the Spark UI as the evidence for the specific run rather than assuming that every ORDER BY will read its input twice.

The additional scan helped explain some of the work. But it was not the number that concerned me most.

The stage metrics also showed approximately 332 GB of shuffle write.

That looked strange when compared with a 131 GB source table. No part of the query was duplicating the rows, yet the shuffle was much larger than the stored dataset.

The two values were not measuring the same physical representation.

The 131 GB value described compressed, columnar Parquet files in the Delta table. The 332 GB value described data serialized into Spark’s shuffle format. Different encodings and compression behaviour can produce very different sizes, even when the logical rows have not changed.

For this investigation, the important point was not why the two representations had identical or different compression ratios. It was that Spark now had approximately 332 GB of serialized shuffle data to distribute across the downstream tasks.

Dividing 332 GB Across 200 Partitions Exposed the Bottleneck

The job was using only 200 shuffle partitions.

I divided the measured shuffle volume by that partition count:

This was only an average. Some tasks could receive less data and others could receive more. But an average of 1.66 GB per task was already large enough to investigate memory pressure.

Each worker had four cores, which meant four tasks could run concurrently. Those tasks were sharing the executor’s execution-memory pool while also needing memory for sort structures, records, and serializer buffers.

The worker had 16 GB of physical memory, but that did not mean one task could use 16 GB. The operating system, Databricks services, executor overhead, native memory, the JVM heap, and the other concurrent tasks all needed a share.

In this cluster, I observed an executor heap of about 8.8 GB per worker. Even that heap was not fully available for sorting.

Why the Memory Calculation Mattered

Spark calculates its unified execution-and-storage region approximately as:

M = (executor heap – reserved memory) × spark.memory.fraction

With the default:

spark.memory.fraction = 0.6

Spark first reserves about 300 MB and then applies the fraction to the remaining heap.

This unified region is shared by execution operations such as shuffles, joins, aggregations, and sorts, and by cached storage blocks.

One setting is commonly misunderstood here. spark.memory.storageFraction = 0.5 does not create a permanent 50-50 split between execution and storage memory. It defines the protected storage region inside unified memory. Execution can use available space and evict storage blocks above the protected region when required.

Based on the observed heap, the unified-memory settings, and four concurrent tasks, I estimated that each task could work with roughly 1.1 to 1.3 GB of execution memory in this run.

That was not a fixed allocation. Spark does not permanently reserve an identical block of memory for each core. The estimate simply gave me the comparison I needed:

• Average shuffle partition: approximately 1.66 GB
• Rough working-memory capacity per concurrent task: approximately 1.1 to 1.3 GB

The task was being asked to sort more data than it could comfortably keep in memory. The next place to look was spill.

The Cluster Looked Healthy Because Spark Was Spilling

When a sort cannot keep all its intermediate data in execution memory, Spark does not immediately fail. It writes part of that data to local disk, frees memory, continues processing, and later reads or merges the spilled data when required.

That behaviour keeps the job alive, but it introduces more work: local disk writes, local disk reads, serialization, deserialization, and additional merge passes during the external sort.

This was why the cluster could look healthy while the job remained slow. The stage stayed green, the executors remained alive, and the tasks continued to progress. The extra time was hiding in disk I/O and merge work.

The Spark UI confirmed it through memoryBytesSpilled and diskBytesSpilled.

The name memoryBytesSpilled can be misleading. It represents the estimated in-memory size of the data that was spilled. diskBytesSpilled represents the bytes actually written to disk.

The stage summary reported hundreds of GiB of spill. One screenshot also showed 0 B in the Min column, which only meant that at least one task did not spill. The median and executor-level metrics told the real story: spill was widespread across the stage.

What Happened When I Made the Tasks Smaller

I could not remove the 332 GB shuffle without changing the requirement for a global sort. What I could change was how many tasks shared that work.

I used 128 MB as a starting target, not as a universal Spark rule:

332 GB × 1,024 = approximately 340,000 MB

340,000 MB / 128 MB = approximately 2,656 partitions

The calculation gave me about 2,656 partitions. I rounded it to 2,700:

spark.conf.set(“spark.sql.shuffle.partitions”, 2700)

I kept the dataset, query, cluster, and execution path the same. Only the initial shuffle-partition count changed.

The result was much clearer than the configuration change itself might suggest.

The observed average partition size fell from approximately 1.66 GB to around 140 MB. Spill reduced sharply, and the runtime dropped from 49 minutes to 27 minutes.

I had not added workers or rewritten the query. I had changed the size of the work assigned to each task.

This is also why I would not copy the value 2,700 into another Spark job without measuring it. A useful partition size depends on the operation, row width, compression, available execution memory, concurrency, storage throughput, network throughput, scheduling overhead, and skew.

Too few partitions create oversized tasks and spill. Too many can create tiny tasks and scheduling overhead. The Spark UI should decide where the balance sits for the workload.

Understanding how shuffle size, execution memory, and disk spill connect is a core part of the hands-on Spark performance tuning scenarios covered in TrendyTech’s Databricks Performance Tuning Program.

Why AQE Didn’t Rescue the Original 200 Partitions

AQE was already enabled in the 49-minute baseline run, which made one question unavoidable: why did it not simply turn 200 oversized partitions into the 2,700 partitions the workload needed?

AQE can coalesce small post-shuffle partitions, and it can split skewed partitions in supported scenarios such as skewed joins. But in this run, it did not expand the undersized global range shuffle from 200 partitions to the higher starting count required for the sort.

That is why the initial value still mattered.

The practical pattern I use is to configure enough initial partitions to avoid oversized shuffle blocks, allow AQE to coalesce them when the runtime data is smaller than expected, and then verify the final sizes in the AQEShuffleRead and stage metrics.

AQE can correct an overestimated starting point by merging small partitions. It does not remove the need to choose a sensible starting point.

Could More Memory Have Solved the Same Problem?

Once I knew that each task was receiving too much data for the available execution memory, adding memory looked like another reasonable option.

I tried increasing:

spark.conf.set(“spark.memory.fraction”, 0.8)

That increased the unified execution-and-storage region inside the executor heap and reduced some memory pressure. But it did not reduce the 332 GB shuffle or make a 1.66 GB partition smaller. Large partitions could still spill.

The extra execution memory became more useful after the tasks themselves became smaller.

There was another reason not to treat spark.memory.fraction as a free performance switch. The remaining heap still supports user objects, metadata, and memory outside Spark’s unified region. Increasing the fraction too aggressively can move the pressure somewhere else.

The lesson from this run was straightforward: fix partition sizing first and use additional execution memory as support.

What I Observed with Larger Workers

I also examined whether a larger VM could solve the problem by giving each task more memory.

It reduced some memory pressure, but it did not produce the best end-to-end runtime. As the worker became larger, the bottleneck began shifting towards cloud reads, shuffle throughput, local-disk throughput, and garbage collection.

A larger worker can help when memory per task is the dominant constraint. It still does not remove a 332 GB global shuffle.

In a separate VM comparison, a general-purpose configuration with 64 GB RAM and 16 cores completed the observed workload in approximately 5.8 minutes, while a memory-optimized configuration with 128 GB RAM and the same 16 cores took approximately 6.5 minutes.

Those timings belonged to that separate comparison and should not be compared directly with the 49-minute baseline. They demonstrated a different point: doubling the memory per core did not automatically make the workload faster once memory was no longer the only bottleneck.

I choose a VM according to the measured bottleneck, not according to the largest memory specification.

What Actually Reduced the Runtime from 49 to 27 Minutes

The 49-minute run connected four measurements:

• Approximately 332 GB of shuffle write
• 200 shuffle partitions
• Approximately 1.66 GB per partition on average
• Roughly 1.1 to 1.3 GB of working memory per concurrent task

More memory helped a task tolerate a larger partition. Increasing the partition count changed the size of the task itself.

For this workload, that distinction mattered. Once I configured 2,700 initial shuffle partitions, the observed task size fell to around 140 MB, spill reduced sharply, and the same job completed in 27 minutes.

The result does not prove that 2,700 is the correct value for every Spark job. It proves that the original 200-partition starting point was too low for this shuffle volume, operation, and cluster.

How I Investigate Spark Data Spill

When a Spark job is slow but the cluster looks healthy, I do not begin by resizing the cluster. I begin with the stage that is consuming the time.

First, I confirm whether it is a shuffle stage by checking the physical plan, shuffle read, shuffle write, and stage duration. Operators such as Exchange, Sort, SortMergeJoin, aggregations, and window functions usually reveal where the data movement begins.

Then I compare the minimum, median, 75th percentile, and maximum task metrics. Looking at an average or the Min column alone can hide skew, spill, or a long-running tail.

I check both memory bytes spilled and disk bytes spilled, estimate the approximate partition size from the shuffle volume and partition count, and compare that estimate with the actual per-task metrics.

Finally, I review executor memory, cores per executor, concurrent tasks, cache usage, and off-heap configuration before changing one variable and running the same workload again.

The comparison must use the same input, query, and cluster conditions. Otherwise, a faster result could come from cache, cluster warm-up, different input, or normal runtime variation rather than the configuration being tested.

Final Takeaway

One global ORDER BY created range sampling, 332 GB of shuffle write, oversized sort tasks, and extensive disk spill.

The query text did not explain the 49-minute runtime. The Spark UI did.

It showed that Spark was trying to distribute 332 GB across only 200 initial partitions. That created tasks averaging approximately 1.66 GB, which was more than the available execution memory could comfortably sort.

Once the initial partition count increased to 2,700, the observed task size fell to around 140 MB, spill reduced sharply, and the workload completed in 27 minutes.

The reusable lesson is not a configuration value.

Open the Spark UI. Follow the shuffle. Estimate the task size. Check spill. Change one variable. Compare the evidence.

That is what Spark performance tuning comes down to: finding where the time is being spent and changing the part of the execution that is actually slow.

If you want to practise this kind of Spark UI investigation through hands-on Spark and Databricks experiments, explore TrendyTech’s Databricks Performance Tuning Program.

Documentation References

1. Apache Spark Tuning Guide: https://spark.apache.org/docs/latest/tuning.html
2. Apache Spark SQL Performance Tuning: https://spark.apache.org/docs/latest/sql-performance-tuning.html
3. Spark configuration for spark.memory.fraction and spark.memory.storageFraction: https://spark.apache.org/docs/latest/configuration.html

Frequently Asked Questions