Reading Shuffle Volume
Check 2 asks one question of the slow stage: does the amount it shuffles match what its operation needs? The stage page header gives 3 totals. Input is what the stage read from storage. Shuffle Write is what its map tasks wrote for the next stage, and Shuffle Read is what its reduce tasks fetched from the stage before. The ratio to compute is a map stage's shuffle write over its input, or for a reduce stage, its shuffle read over the input of the stages feeding it.
That ratio has no single healthy value. Each operation shape has an expected value, and the verdict comes from comparing the measured ratio with it:
| Operation | Expected write over input | Why |
|---|---|---|
| groupBy on a low-cardinality key | Well under 0.01 | Each map task combines its rows to one per key before writing |
| groupBy on a near-unique key | Close to 1 | Combining finds almost no repeated keys inside a map task |
| Shuffle join or window | About 0.5 to 1.5 | Every kept row crosses; columnar input and shuffle blocks compress differently |
| Broadcast join | 0 for the large side | The small side ships whole and the large side stays put |
| After an explode or a duplicate-key join | Above 1.5 | Rows multiplied before the write |
Count records, then bytes
Bytes depend on encoding, so count records instead. The stage page reports Shuffle Write Size / Records, and an aggregation that combines has a hard ceiling on the records it can write: each map task writes at most one row per distinct key it saw. Stage 3 of the ranking job ran 2,800 map tasks over viewing_events, which covers 190 countries. A groupBy on country over that scan can write no more records than this, whether the table holds 9 billion rows or 90 billion:
- 532,000
The animation runs that combine on 15 rows of viewing minutes split into 3 partitions. Watch the shuffle: each map task sends at most 3 rows, one per country, so 15 input rows become no more than 9 shuffled ones.
views.groupBy("country").sum("minutes").collect()| country | minutes |
|---|---|
| US | 42 |
| BR | 15 |
| US | 30 |
| IN | 22 |
| US | 8 |
| BR | 61 |
| IN | 12 |
| US | 25 |
| BR | 9 |
| IN | 40 |
| US | 18 |
| US | 33 |
| IN | 5 |
| BR | 27 |
| US | 11 |
Same scan, 3 verdicts
Each shuffle write below is measured against the same 380 GB scan of viewing events. The first 2 come from the scan stage itself. The third comes from the join's reduce stage, where the joined rows enter the window's exchange.
| Operation | Shuffle write | Ratio | Verdict |
|---|---|---|---|
| groupBy country, sum minutes | 15 MB | 0.00004 | Healthy: combining collapsed each task to 190 rows at most |
| groupBy profile_id, collect_list of titles | 361 GB | 0.95 | Expected for the key: profiles barely repeat inside a task |
| window by title_id after a join to device_history on device_id | 1.9 TB | 5.0 | Excessive: device_id repeats on both sides, so the join multiplied rows before the window's exchange |
The second row deserves a closer look even though it reads as expected. Grouping by profile sends nearly every row across the network, so the question becomes whether the job needs profile-level lists at all. An expected ratio tells you the shuffle matches the code; whether the code should shuffle at that grain is a separate decision.
Where an excessive shuffle comes from
- Rows multiplied before the write: an explode shows Shuffle Write records above Input records in its own stage. A join on a key that repeats on both sides shows it one stage later: the join's map stages write their input rows, and the stage taking the join's output into the next exchange writes far more records than the scan read.
- A broadcast that did not happen: the small side's estimated size sat above spark.sql.autoBroadcastJoinThreshold, so the large side crossed the network too. The plan shows SortMergeJoin where you expected BroadcastHashJoin.
- The same data shuffled twice on one key: 2 Exchange nodes with identical hash partitioning in a single plan, where the first shuffle's output could have been reused.
Input counts bytes read after Parquet column pruning, so a query touching 4 of 60 columns shows only those 4 columns' bytes. Measuring shuffle write against the full table size on disk makes every shuffle look healthier than it is.
A stage scans 120 GB and runs a sum of revenue grouped by store_id, with 1,000 map tasks over 500 stores. Its Shuffle Write reads 96 GB. What is the verdict?
sales.groupBy("store_id").sum("revenue")- Excessive: 1,000 tasks and 500 stores cap the write at 500,000 records, so 96 GB means rows crossed uncombined.
- Healthy: a groupBy shuffles every row to the task that owns its key, so a write near the input is expected.
- Excessive: 500 stores is too few keys, so the shuffle should be split across more partitions to cut the write.
- Healthy: a ratio of 0.8 sits inside the 0.5 to 1.5 band, the expected range for most shuffles.
About this lesson
This section is part of the The Spark UI: Intermediate 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.