Sort-Based Shuffle Internals
A shuffle has 2 halves joined by files on local disk. Map tasks, the last tasks of the stage before the boundary, write those files. Reduce tasks, the first tasks of the stage after it, read them. The layout inside those files decides how many disk operations, open file handles and network requests the shuffle costs.
Each map task writes a data file and an index
Since Spark 2.0, the sort-based shuffle manager is the only one Spark ships. A map task with output for R reducers writes 2 files. The data file holds all of its output rows as R segments laid end to end in reducer order. The index file holds R + 1 byte offsets: where each segment starts, plus the end of the file. Reducer 3 reads from offset 3 to offset 4, and those bytes are its block from this map task.
| Reducer | Reads from byte | Reads to byte | Block size |
|---|---|---|---|
| 0 | 0 | 18,400 | 18,400 |
| 1 | 18,400 | 18,400 | 0 (no rows) |
| 2 | 18,400 | 51,200 | 32,800 |
| 3 | 51,200 | 60,000 | 8,800 |
This map task's index file holds 5 numbers for 4 reducers: 0, 18,400, 18,400, 51,200 and 60,000. Reducer 1 has an empty block, recorded as 2 equal offsets. The driver also knows each block's size from the map task's status report, so reducer 1 never requests its empty block.
The old design: a file for every block
Spark's first shuffle wrote a separate file for every pair of map task and reducer. Sort-based shuffle became the default in Spark 1.2, and Spark 2.0 removed the old design.
| Layout | Files after 1,000 map tasks feed 1,000 reducers | Logical blocks |
|---|---|---|
| One file per map task and reducer | 1,000,000 | 1,000,000 |
| Sort-based: data file plus index | 2,000 | 1,000,000 |
The block count stays at 1 million either way, because each block became a byte range inside a data file instead of a file of its own. The sort layout removes the million file creations, the inode pressure and the per-node limit on open file handles. In exchange, every map task must order its output by reducer before writing it.
Inside one map task
The sorting writers, step by step
- Assign: The partitioner turns each row's key into a reducer id from 0 to R minus 1: a hash for groupBy, a sampled range for orderBy.
- Buffer: Rows collect in a memory buffer, each tagged with its reducer id.
- Sort and spill: When the buffer reaches the memory the task was granted, the task sorts it by reducer id and writes it to a spill file.
- Merge: The task merges its spill files and what remains in memory into the data file, segment by segment, then writes the index.
The sort is by reducer id. Inside a segment, rows carry no key order unless the operation asked for one. A task that never fills its buffer makes a single pass. A task that spills writes every spilled row twice, once to a spill file and once into the merged data file, and reads it back in between. Spill (Disk) on the stage page measures that extra traffic.
3 writers produce 1 layout
Spark picks 1 of 3 writers for each shuffle when it registers it. All 3 produce the same data file and index, so reducers never know which one ran.
| Writer | Spark picks it when | How it builds the data file |
|---|---|---|
| Bypass | the writer does no combining and R is at most `spark.shuffle.sort.bypassMergeThreshold`, 200 by default | opens R files at once, 1 per reducer, then concatenates them in order |
| Serialized | the writer does no combining, the serializer can move records as raw bytes, and R is at most 16,777,216 | sorts 8-byte pointers that pack the reducer id with the record's address, so rows stay as bytes |
| Base sort | the writer itself must combine values per key | sorts row objects and merges values that share a key as it goes |
DataFrame aggregations do their combining in the partial aggregate before the shuffle, so their shuffle writer never combines. That leaves R as the deciding number. With spark.sql.shuffle.partitions at its default of 200, R equals the threshold, and a default DataFrame shuffle takes the bypass path: each map task holds 200 files open while it writes. Set 201 or more and the same job switches to the serialized writer, with no change to its output.
The reduce side
Before fetching, a reduce task asks the driver where every map task's output lives and how big each block is. It then pulls its block from each map task: from local disk when the map ran on its own executor, over the network otherwise. spark.reducer.maxSizeInFlight, 48MB by default, limits how many bytes it has requested and not yet received. If the operation needs key order, the reducer sorts what arrives and spills when it runs short of memory.
A DataFrame groupBy runs with 400 map tasks and spark.sql.shuffle.partitions set to 300. When the map stage finishes, how many shuffle files sit on local disks, and which writer wrote them?
- 120,000 files, 1 per map task and reducer pair, written by the bypass writer
- 800 files written by the bypass writer, because the threshold compares against map tasks
- 800 files, a data file and an index per map task, written by the serialized writer
- 700 files, a data file per map task plus an index per reducer, from the serialized writer
About this lesson
This section is part of the Shuffles: Advanced 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.