The Shuffle Write
A shuffle runs in 2 halves, split across a stage boundary. The first half, the shuffle write, is the last thing every task does in the stage before the wide transformation. Each of those tasks, called a map task, runs its pipelined narrow work, decides for every output row which downstream partition owns it, and writes the rows to its executor's local disk, grouped by that partition. The stage finishes only when the last map task has written.
plays = spark.table("plays")
streams = (plays
.filter(F.col("ms_played") > 30000)
.groupBy("listener_id", "artist_id")
.agg(F.count("*").alias("streams")))
streams.write.parquet("/out/streams")The rest of this lesson follows that job over a year of plays read as 400 partitions on a cluster of 20 executors. Stage 0 runs 400 map tasks. Each reads its split, applies the filter in the same pass, and for every surviving row works out which of the 200 reduce partitions owns its (listener_id, artist_id) pair. The 200 is the default of spark.sql.shuffle.partitions, the setting that fixes how many partitions a DataFrame shuffle produces.
Choosing the target partition
For a groupBy, the owner comes from a hash of the grouping columns, taken modulo the number of reduce partitions. Every map task applies the same function, so a given listener and artist pair goes to the same reducer whichever task happened to read it. That agreement lets reducer 17 collect every row for its keys without asking anyone where they went. A later section of this lesson opens up the hash and the other rule Spark uses.
What lands on disk
Each map task writes 2 files. The data file holds all its output rows laid out as 200 consecutive runs, 1 per reduce partition, and the index file records the byte offset where each run starts, so any run can be served without scanning the rest. A single run, the rows a map task holds for a single reducer, is a shuffle block. Rows are serialized and compressed on the way out, so they usually take far fewer bytes on disk than they did in memory.
| Quantity | Rule | Running example |
|---|---|---|
| Map tasks | 1 per input partition | 400 |
| Reduce partitions | `spark.sql.shuffle.partitions` | 200 |
| Files on local disk | 2 per map task | 800 |
| Shuffle blocks | map tasks times reduce partitions | 80,000 |
| Blocks per map task | 1 per reduce partition | 200 |
Files grow with map tasks alone; blocks grow with the product of both sides. Spark has kept 1 data file per map task since version 1.2, when sort-based shuffle replaced a layout that opened a separate file for every block.
Which disk, and for how long
The files go under spark.local.dir, which is /tmp unless set, on the machine running the executor; cluster managers often substitute directories of their own. A job writing 500GB of shuffle across 20 executors therefore needs about 25GB of free local disk on each, and when one machine fills up, its map tasks fail with No space left on device while the cluster's main storage sits half empty. The files also outlive the tasks that wrote them, because the next stage still has to fetch them.
The animation shrinks the job to 9 orders read as 2 partitions and counted per region into 2 reduce partitions. Watch the end of the first stage: each map task arranges its output by target partition and writes it before any reduce task starts. The final counts are US 3, EU 3, APAC 2 and LATAM 1.
orders.groupBy("region").count().collect()| order_id | region |
|---|---|
| 1 | US |
| 2 | EU |
| 3 | US |
| 4 | APAC |
| 5 | EU |
| 6 | LATAM |
| 7 | US |
| 8 | APAC |
| 9 | EU |
The write in the Spark UI
On the Stages tab, each stage's row carries a Shuffle Write column: the bytes and records all its tasks wrote. Open the map stage and every task lists its own Shuffle Write Size / Records and its shuffle write time. Read them against Input. A write far larger than the input means rows multiplied before the shuffle, and a write time that eats a large share of each task points at slow or crowded local disks.
A stage of 300 map tasks feeds a groupBy with spark.sql.shuffle.partitions set to 100. What does its shuffle write leave on the executors' local disks?
- 30,000 files, 1 per block, since each map task opens a file for every reducer
- 600 files holding 30,000 blocks, since each map task writes a data file and an index file
- 400 files holding 30,000 blocks, since there is 1 file per map task and 1 per reducer
- 600 files holding 400 blocks, since there is 1 block per map task and 1 per reducer
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.