Spark Shuffle Explained: What It Costs, How to Read It in the UI, and the Rewrites That Remove It

Every Spark job I have ever been asked to make faster was slow for the same reason. Not bad code, not undersized clusters. Data moving across the network when it did not have to.

A shuffle is the only operation in Spark that hits disk, network, and CPU serialization in one step. A 2 TB shuffle on a 40-executor cluster is roughly 2 TB written to local SSD, 2 TB read back, and 2 TB pushed through the network in a fan-out pattern that gets worse as you add executors. I have seen a job go from 55 minutes to 6 by removing two shuffles and changing nothing else — same cluster, same data, same output.

The tricky part is that shuffles are invisible in the code you write. groupBy does not say “write 2 TB to disk.” join does not warn you that both sides are about to be repartitioned by the join key. You find out from the Spark UI, after the fact, in a stage that has been running for 40 minutes.

So let us make it visible: what actually happens during a shuffle, what it costs, how to read it in the UI, and the specific rewrites that remove it.

What actually happens during a shuffle

Spark splits a job into stages at every point where data must be redistributed. Inside a stage, every task is a pipeline of narrow transformations on one partition — no coordination needed. At a stage boundary, tasks must exchange data by key, and that exchange is the shuffle.

Mechanically it happens in two halves:

Map side. Each task in the upstream stage takes its output rows, computes hash(key) % numPartitions for each one, sorts records into per-target-partition buckets, and writes them to local disk as one data file plus an index file. If the sort buffer fills, the task spills sorted runs to disk and merges them. Nothing is sent yet; everything is materialized on the executor’s local volume.

Reduce side. Each task in the downstream stage asks every upstream executor for its slice of that partition. With 400 map tasks and 400 reduce tasks, that is 160,000 fetch requests. The fetched blocks are deserialized, merged, and fed to the aggregation or join.

flowchart TD
  m1["Map task 1"] --> w1["Sort, spill, write local files"]
  m2["Map task 2"] --> w2["Sort, spill, write local files"]
  m3["Map task 3"] --> w3["Sort, spill, write local files"]
  w1 --> net["Network fetch: M x R requests"]
  w2 --> net
  w3 --> net
  net --> r1["Reduce task 1"]
  net --> r2["Reduce task 2"]
  r1 --> out["Stage output"]
  r2 --> out

Every byte is written to disk, requested over the network, and deserialized. Three costs, one operation.

Two consequences fall straight out of that picture. First, shuffle files live on the executor that wrote them, which is why losing an executor mid-job triggers recomputation of its map output — and why the external shuffle service exists. Second, the number of fetch requests grows as the product of both stage widths, so throwing executors at a shuffle-bound job can make it slower.

What it costs, in numbers I have measured

On a typical cloud cluster with local NVMe and 10 Gbps networking, I budget roughly:

• 250-400 MB/s per executor of effective shuffle write throughput once serialization is included.
• 1.5-3x the logical data size written to disk, because Java object serialization on wide rows is not compact and spill duplicates data.
• Network as the ceiling past about 20 executors. A 2 TB shuffle across 10 Gbps links takes minutes of pure wire time even if nothing else goes wrong.

The number I actually watch is shuffle bytes per output row. A daily aggregation that reads 800 GB and emits 40 million rows should not be shuffling 800 GB. If it is, the aggregation is happening after the exchange instead of before it, and that is a plan problem I can fix in one line.

Reading a shuffle in the Spark UI

Open the SQL tab, click the query, and read the plan bottom-up. Every Exchange node is a shuffle. Every Exchange hashpartitioning is a shuffle on a key; Exchange SinglePartition is worse, because it funnels everything through one task.

Then go to the Stages tab and read four columns:

• Shuffle Write — bytes the upstream stage materialized. This is your network bill.
• Shuffle Read — bytes the downstream stage fetched. Should be close to write; a large gap means skew or a filter pushed past the exchange.
• Shuffle Spill (Memory) and Shuffle Spill (Disk) — data that did not fit in execution memory. Disk spill greater than shuffle write means every byte moved at least twice.
• Task duration distribution — click into the stage. If max is 10x median, you have skew, not a shuffle problem, and the fix is in the data skew playbook.

# Programmatic version of what the UI shows, for CI regression checks
from pyspark.sql import functions as F

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

agg = (
    df.groupBy("customer_id")
      .agg(F.sum("net_amount").alias("lifetime_value"))
)

# The plan is the source of truth. Count the Exchange nodes.
agg.explain("formatted")

# Metrics after the fact, per stage, from the listener bus
for s in spark.sparkContext.statusTracker().getStageInfo(0) or []:
    print(s)

I keep explain("formatted") output in code review for any query touching more than a terabyte. A PR that adds an Exchange should say why. That habit alone catches most accidental shuffles before they reach production, and it costs a reviewer about a minute.

The rewrites that remove shuffles

Three changes account for almost every shuffle I have deleted from a production job. None of them require a config change.

Broadcast the small side

The highest-return change of the three. A join between a 3 TB fact table and a 40 MB dimension does not need a shuffle at all. Ship the dimension to every executor and join locally.

from pyspark.sql import functions as F

fact = spark.read.table("prod.silver.orders")          # 3 TB
dim  = spark.read.table("prod.gold.dim_customer")      # 38 MB

# Explicit is better than hoping the optimizer has good stats
joined = fact.join(F.broadcast(dim), "customer_id", "left")

Spark auto-broadcasts below spark.sql.autoBroadcastJoinThreshold, default 10 MB. That default is far too conservative for modern executor memory. I run 64 MB on production clusters with 32 GB executors and set it higher per-job when I know the dimension size:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 64 * 1024 * 1024)

The failure mode is real, though: broadcasting a table that turns out to be 4 GB gives you a driver OOM during collect, or executor OOM after. Broadcast when you know the size, and know it from DESCRIBE DETAIL or table statistics rather than a guess.

Aggregate before you join

This is the one that produced my 55-to-6-minute win. The original query joined a wide events table to a dimension, then grouped. The join shuffled 2.1 TB. Grouping first shuffled 60 GB, and the join after that was a broadcast.

-- Before: shuffles the full events table into the join
SELECT c.segment, count(*) AS events
FROM prod.silver.events e
JOIN prod.gold.dim_customer c USING (customer_id)
WHERE e.event_date >= current_date() - INTERVAL 7 DAYS
GROUP BY c.segment;

-- After: pre-aggregate to one row per customer, then broadcast join
WITH per_customer AS (
  SELECT customer_id, count(*) AS events
  FROM prod.silver.events
  WHERE event_date >= current_date() - INTERVAL 7 DAYS
  GROUP BY customer_id          -- shuffles 60 GB, not 2.1 TB
)
SELECT c.segment, sum(p.events) AS events
FROM per_customer p
JOIN prod.gold.dim_customer c USING (customer_id)
GROUP BY c.segment;

Both queries return the same numbers. One moves 35x less data. Spark’s optimizer will not do this for you, because pushing an aggregation through a join is only valid under conditions it cannot always prove. This is the same class of rewrite as the ones in SQL anti-patterns: the optimizer is good, but it is not going to restructure your semantics.

Stop shuffling on purpose

Half the repartition calls I find in production code are cargo cult. Some rules I apply:

• repartition(n) is a full shuffle. Use it when you genuinely need to change parallelism upward or fix skew, not “to spread the data out” before a write.
• coalesce(n) does not shuffle, but it also reduces upstream parallelism, so coalesce(1) before a write can serialize your whole job.
• df.distinct() on a wide table shuffles every column. Select the keys first.
• orderBy on the full result is a SinglePartition exchange unless you only need per-partition ordering, in which case sortWithinPartitions is free.
• Writing to a partitioned table with partitionBy triggers a shuffle if you also enable optimized writes. That one is usually worth it, because the alternative is the small-file problem from the Delta transaction log post.

Tuning the shuffles you cannot delete

Some shuffles are load-bearing. For those, the goal is partition sizing: aim for 128-200 MB of shuffle data per task. Too large and you spill; too small and scheduling overhead dominates.

Adaptive Query Execution handles most of this at runtime, and it has been on by default since Spark 3.2. Keep it on:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

# Only relevant when AQE is off or for the initial pre-AQE partition count
spark.conf.set("spark.sql.shuffle.partitions", "auto")

AQE reads the map-output statistics and coalesces small partitions into right-sized ones, converts sort-merge joins to broadcast joins when a side turns out small, and splits skewed partitions. What it cannot do is remove an exchange that the plan requires. Tune after you rewrite, never instead.

For very large shuffles, two more knobs earn their keep: spark.shuffle.file.buffer at 1 MB (from 32 KB) reduces syscalls on the map side, and spark.reducer.maxSizeInFlight at 96 MB (from 48 MB) helps when the network is not saturated. Both trade memory for throughput, so validate against the spill numbers from the memory post.

Pitfalls

Adding executors to a shuffle-bound job. The fetch request count scales with map tasks times reduce tasks. Doubling parallelism on a network-bound stage can extend runtime. Measure before you scale.

Trusting auto-broadcast with stale statistics. If the table has never had ANALYZE TABLE ... COMPUTE STATISTICS run and is not a Delta table with recent stats, Spark estimates size from file bytes and gets it wrong on compressed columnar data. Use an explicit broadcast hint for joins you care about.

coalesce(1) to get one output file. It removes parallelism from every upstream stage in the same stage boundary, not just the write. Use repartition(1) if you truly need a single file, and accept the shuffle, or better, compact afterward.

Caching before a shuffle to “help”. Cache stores the pre-shuffle data, which does nothing for exchange cost and steals execution memory that the shuffle sort needed. This causes spill that was not there before.

Ignoring spill because the job still finishes. Disk spill larger than shuffle write means you are paying for the same bytes three times. It is the loudest signal in the UI and the one teams most reliably scroll past.

Assuming AQE fixes skew. It splits skewed shuffle partitions only when it can detect them from map statistics and only above configured thresholds. A single 200 GB key still lands in one task.

FAQ

How many shuffle partitions should I set?

With AQE on, set advisoryPartitionSizeInBytes to 128 MB and stop thinking about the count. Without AQE, target total shuffle bytes divided by 128 MB, rounded to a multiple of your total core count. The old default of 200 is wrong for almost every real dataset.

Is a sort-merge join always worse than a broadcast join?

No. Broadcast requires the small side to fit in every executor’s memory plus the driver’s. Above a few hundred megabytes, or when both sides are large, sort-merge on well-sized partitions is the right plan. Broadcast is a win at extreme size ratios, which happens to be most fact-to-dimension joins.

Does bucketing actually remove shuffles?

Yes, when both tables are bucketed on the join key with the same bucket count and read through the metastore. In practice I rarely use it: it locks the physical layout, breaks when bucket counts drift, and Delta’s liquid clustering plus AQE covers most of what people wanted from it.

Why is my Shuffle Read much smaller than Shuffle Write?

Usually a filter or aggregation that AQE pushed down after the exchange was planned, or a join that eliminated rows. It is generally good news. The inverse — read much larger than write on a single task — is skew.

Does the external shuffle service still matter?

On dynamic-allocation clusters, yes: it lets Spark release executors without losing their shuffle files. On Databricks and recent Spark versions, push-based shuffle and disaggregated shuffle storage have made this less manual, but if you run dynamic allocation on YARN or Kubernetes without it, scale-down triggers recomputation.

What this means for your pipelines

Treat shuffle bytes as a first-class metric, not a curiosity. For every scheduled job over a terabyte, I record shuffle write, disk spill, and wall clock per run, and I alert on shuffle bytes growing faster than input bytes. That ratio drifting upward is almost always someone adding a join or a distinct in a place where the data has not been reduced yet.

The rewrites are not exotic. Broadcast the small side. Aggregate before joining. Delete the repartition nobody can justify. Sort within partitions when global order is not required. Each of these is a few lines, and each removes an entire disk-plus-network round trip for your whole dataset.

Then, and only then, tune. AQE with a 128 MB advisory size handles partition sizing better than you will by hand, and the remaining knobs are worth a few percent. The plan is worth an order of magnitude. Read the Exchange nodes before you touch a config file, and pair that habit with the PySpark fundamentals your whole team already shares.