The Shuffle Read
The reduce stage waits for the whole map stage, since any map task might hold rows for any reducer. As each map task finishes, it reports to the driver where its data file lives and the size of each of its blocks. When all 400 have reported, the driver has a complete map of the shuffle, and the 200 reduce tasks can start.
Reduce task 17 then fetches block 17 from each of the 400 data files. Blocks written on its own executor are read from local disk; the rest arrive over the network from the other 19 executors. With map tasks spread evenly across the cluster, roughly 1 block in 20 is local and 19 in 20 are remote, so nearly every shuffled byte crosses the network once.
- Read plays (source, parquet input)
- Map tasks (transform, stage 0); parallelism: 400 partitions
- Shuffle files (storage, spark.local.dir)
- Reduce tasks (transform, stage 1); parallelism: 200 partitions
- Write streams (storage, parquet output)
- Read plays -> Map tasks
- Map tasks -> Shuffle files
- Shuffle files -> Reduce tasks
- Reduce tasks -> Write streams
From blocks to results
Fetches go out in parallel batches, so a reducer pulls from several executors at once while it processes blocks that have already arrived. For the groupBy, processing means merging: the reducer feeds each incoming row into a hash map keyed by (listener_id, artist_id), adding counts for rows that share a key. Once the last block is merged, every key's count is final, because no other reducer holds rows for those keys. The finished rows then flow into the narrow work after the shuffle in the same stage, here the parquet write.
The read mirrors the write
Each byte stage 0 writes, stage 1 reads, so the reduce stage's Shuffle Read total matches the map stage's Shuffle Write total. The split differs. Writes were divided by input partition, and reads are divided by key, each reducer taking whatever share of the data its keys own. In this job that share stays close across reducers. As the next section shows, each map task sends no more than 1 row for each (listener_id, artist_id) pair, so a listener who streams all day adds at most 400 rows per artist, and the hash spreads their artists over many reducers. Read sizes split apart when raw rows cross the shuffle, as after repartition("listener_id"): every play of a heavy listener lands on 1 reducer, that reducer reads far more than the median, and the stage waits on it.
Read metrics in the Spark UI
| Task metric | What it counts | What a high value suggests |
|---|---|---|
| Shuffle Read Size / Records | Bytes and rows the task fetched | Far above the median: its keys own a large share of rows |
| Shuffle Remote Reads | Bytes fetched from other executors | Near the total: expected when executors are many |
| Shuffle Read Blocked Time | Time the task waited with no block to process | A large share of task time: fetching is the bottleneck |
A reduce stage of 200 tasks is slow. Read sizes per task are within 10% of each other, the median task takes 40 seconds, and the median Shuffle Read Blocked Time is 31 seconds. Where should you look first?
- At fetching: tasks spend most of their time waiting for blocks from other executors
- At skew: a few reducers own most of the keys and hold up the whole stage
- At the map stage: too few map tasks wrote the files the reducers need
- At the merge: feeding rows into the hash map takes most of each task
Reducers with nothing to read
The plan sets the reduce task count to spark.sql.shuffle.partitions, whatever the data holds. With adaptive execution turned off (spark.sql.adaptive.enabled set to false), a groupBy on a column with 3 distinct values launches all 200 reduce tasks. Each key's rows land in a single partition, so at least 197 tasks find every block empty and finish with no output, more if 2 of the keys hash to the same partition. Each empty task still costs the driver a scheduling round trip. With adaptive execution on, the default since Spark 3.2, Spark measures the map output once the map stage ends and merges the small and empty partitions into a few tasks, often 1 for 3 keys. The Adaptive Query Execution lessons cover how.
The animation sets the reduce count to 6 and sums 6 orders spread over 3 regions. Before you play it, predict how many of the 6 reduce tasks receive any rows. The totals it returns are US 260, EU 110 and APAC 60.
spark.conf.set("spark.sql.shuffle.partitions", 6)
(orders
.groupBy("region")
.agg(F.sum("amount").alias("revenue"))
.collect())| order_id | region | amount |
|---|---|---|
| 1 | US | 120 |
| 2 | EU | 80 |
| 3 | US | 45 |
| 4 | APAC | 60 |
| 5 | EU | 30 |
| 6 | US | 95 |
events is read as 120 partitions, and its country column holds 8 distinct values. If spark.sql.shuffle.partitions is 50, how many shuffle blocks belong to each reduce task of events.groupBy("country").count(), and how many reduce tasks can produce rows?
- 50 blocks each, and at most 8 tasks produce rows
- 120 blocks each, and all 50 tasks produce rows
- 8 blocks each, and at most 8 tasks produce rows
- 120 blocks each, and at most 8 tasks produce rows
About this lesson
This section is part of the Shuffles: 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.