What a Shuffle Actually Is

A shuffle is an exchange in 2 parts. First, every task that holds input computes a destination for each of its rows from the row's key, and writes the rows into a file per destination on its executor's local disk. Second, a new set of tasks starts, one task per destination, and each fetches its file from every executor over the network, then does its work on the rows it now holds. Spark calls the 2 halves the shuffle write and the shuffle read.

Picking a destination from the key

The destination is computed, never chosen by hand. Spark hashes the key and takes the remainder when divided by the number of output partitions. Hash the string US and you get the same number on every executor, so every US row, whatever partition it started in, is sent to the same destination. Different keys can share a destination, and that causes no trouble: the receiving task sorts them into separate groups when they arrive. Everything else in a shuffle depends on each key having exactly one destination.

destination = hash(key) % num_shuffle_partitions

Rule	Consequence
Equal keys hash to equal values	All rows for a key meet in one reading task
Hash taken modulo the output partition count	Destination is between 0 and n - 1
Several keys can share a destination	A reading task may hold many groups
Every row is routed, even a key seen once	A key appearing a single time still travels

(events
   .groupBy("country")
   .count()
   .collect())

events
event_id	country
1	US
2	DE
3	US
4	BR
5	DE
6	US
7	JP
8	BR
9	US

The diagram has 3 writing tasks and 3 reading tasks. Each writer splits its 3 rows into piles, one pile per reader that has rows coming, and each reader collects its pile from every writer. Watch the US rows: 4 of them, spread across several input partitions, converge on a single reading task, which then counts 4. The reader that receives JP counts a single row, and it still waited for every writer to finish, because any writer might have been holding another JP.

The write side and the read side

Each writer leaves a file per reader on its own executor's disk. Each reader pulls its file from every writer over the network, then aggregates what it fetched.

Writer A (transform, shuffle write)
Writer B (transform, shuffle write)
Writer C (transform, shuffle write)
Shuffle files (storage, local disk)
Reader 0 (transform, shuffle read)
Reader 1 (transform, shuffle read)
Reader 2 (transform, shuffle read)
Driver (consumer, collect())

Writer A -> Shuffle files
Writer B -> Shuffle files
Writer C -> Shuffle files
Shuffle files -> Reader 0
Shuffle files -> Reader 1
Shuffle files -> Reader 2
Reader 0 -> Driver
Reader 1 -> Driver
Reader 2 -> Driver

Write side: Writer A, Writer B, Writer C
Disk: Shuffle files
Read side: Reader 0, Reader 1, Reader 2
Result: Driver

What each half does to a row

Shuffle write: The row is serialized to bytes and appended to the file for its destination on local disk.
The barrier: No reader starts until the last writer has finished, since that writer may hold rows for any reader.
Shuffle read: The reader opens a connection to every executor that wrote and fetches its bytes, then deserializes them back into rows.

The barrier matters as much as the bytes. No reading task starts until every writing task has finished, because the final writer might hold rows for any destination. Spark therefore splits the job at the shuffle: the part before it runs to completion, its files sit on the writers' local disks, and the part after it begins only then. They stay there after the readers finish, so a later action on the same DataFrame reads them instead of rewriting them, and Spark deletes them once nothing refers to that shuffle.

How many readers there are

The number of destinations comes from a setting named spark.sql.shuffle.partitions, whose default is 200. It is independent of how many partitions were read and of how many distinct keys exist. A table read as 8 partitions and grouped by region with 4 distinct regions is split by every writer into 200 destinations, and at most 4 of them receive rows. By default Spark then merges small and empty destinations, so far fewer reading tasks run, and the Adaptive Query Execution lessons cover how. Either way, the input partition count stops mattering once a shuffle happens, because the reading side has a count of its own.

40,000

A groupBy("user_id") shuffles a table that Spark read as 8 partitions. Which statement about where rows land is true?

All rows from input partition 3 land in reading task 3, so the original layout of the table is preserved
Rows with user_id 42 are spread evenly across the reading tasks so that the counting work is balanced
Each reading task receives exactly one user_id, so there are as many readers as there are distinct users
All rows with user_id 42 land in one reading task, whichever of the 8 input partitions they came from

Run the same shape of program on the practice data. Count the rows it returns: that is the number of distinct countries, and so the number of destinations, out of 200, that received any rows.

About this lesson

This section is part of the Shuffles: Beginner 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.