They Ask Why Shuffles Are Costly
Concepts covered: Shuffle Optimization
Once the walk-through has placed the shuffles, the next question is close to predictable: "why are shuffles so expensive?" The thin answer is "because data goes over the network", which names 1 cost out of several and gives no way to tell which dominates. The answer the interviewer wants follows a shuffled byte from the task that produces it to the task that consumes it, names the cost at each step, and says where in the Spark UI each cost shows up.
Follow a byte through the shuffle
| Step | What it costs | Where the Spark UI shows it |
|---|---|---|
| Map-side write | Each row is serialized, grouped by destination partition, compressed with LZ4 and written to local disk with an index file | Shuffle Write Size / Records on the map stage |
| Barrier | The reduce stage starts only after every map task has finished, so the slowest map task sets the start time | The stage page's Event Timeline, a long last task |
| Fetch | Each reducer pulls its block from every map output over the network, up to map tasks times reducers in total | Shuffle Read Blocked Time, Shuffle Remote Reads |
| Merge | Blocks are decompressed, deserialized, then sorted or hashed; whatever does not fit in memory spills | Spill (Memory), Spill (Disk) |
| Skew | A hot key sends its whole volume to a single reducer | Max far above Median in Summary Metrics |
| Lost output | Shuffle files live on the machine that wrote them; losing them reruns the map tasks that produced them | FetchFailed in a task error, the map stage listed again as a retry |
The costs fall into 2 kinds. Writing, fetching and merging scale with bytes, so they dominate a shuffle like the window job's 400GB. The barrier and skew scale with the slowest task, so they dominate whenever a partition is far larger than the rest, whatever the total. Saying which kind you expect for the job in front of you is the second sentence of a good answer.
Put figures on it
The window job gives every cost a figure. The map stage writes about 400GB of compressed shuffle files to local disk, and the reduce stage fetches the same volume back over the network. At 2000 reducers that fetch is 8 million blocks of about 52KB, each a separate request. The reduce stage waits for all 4000 map tasks, so a single task stuck on a slow file leaves most of the 200 cores idle at the tail of the stage. If a fleet account has millions of trips, the reducer holding that rider_id sorts them alone while the rest finish.
The 30-second answer
See both inputs move
This run joins 9 trips to 3 cities with the broadcast threshold set to -1, which forces a sort-merge join. Both tables are read as 3 partitions and both are hashed on city_id into 3 reducers, so every trip and every city row crosses the boundary before a single pair is matched.
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
joined = trips.join(cities, "city_id")
joined.collect()| trip_id | city_id | fare |
|---|---|---|
| 1 | 10 | 12.5 |
| 2 | 20 | 8 |
| 3 | 10 | 4 |
| 4 | 30 | 22 |
| 5 | 20 | 15.5 |
| 6 | 10 | 9 |
| 7 | 30 | 3.5 |
| 8 | 20 | 11 |
| 9 | 10 | 18 |
| city_id | city_name |
|---|---|
| 10 | SF |
| 20 | NYC |
| 30 | LA |
Delete the spark.conf.set line and the plan changes shape: cities, a few hundred bytes against a 10MB threshold, becomes a BroadcastExchange, and trips is joined inside its own scan tasks without crossing the network. That contrast is the strongest answer to "how would you cut shuffle cost?": the cheapest shuffle is the one the plan never contains.
When they push further
The usual next question is how you would reduce the cost. Each lever below removes bytes or a slow task from a row of the table above.
- Filter and project before the shuffle so fewer bytes reach the map-side write.
- Broadcast a side under 10MB, or add cities.hint("broadcast") when the size estimate is missing or wrong.
- Favor aggregates that shrink on the map side, such as sum and count; collect_list ships every value.
- If a join or repartition you needed already hash-partitioned the data on a key, a groupBy on that key adds no shuffle.
- Size shuffle partitions toward 100MB to 200MB each instead of keeping the 200 default.
- Salt a hot key across several reducers when its task's input dwarfs the median.
About this lesson
This section is part of the The Architecture Recital lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.