Why Wide Costs More Than Narrow
A narrow task spends its time on a single resource: the CPU core it runs on, reading rows that are already in memory or streaming in from storage. A wide transformation adds 4 more. It serializes rows to bytes and writes those bytes to local disk; then it sends them across the network, where the far side reads and deserializes them. Each of those is slower than a core comparing 2 values, and a shuffle pays all 4 for every row that crosses.
| Step | Narrow filter | Wide orderBy |
|---|---|---|
| Rows compared or hashed | Every row, in memory | Every row, in memory |
| Rows serialized to bytes | None | Every row alive at the orderBy |
| Bytes written to local disk | None | The whole shuffled dataset |
| Bytes sent over the network | None | The whole shuffled dataset |
| Bytes read back and deserialized | None | The whole shuffled dataset |
| Tasks waiting on other tasks | None | Every reader waits for every writer |
Put sizes on it
Say a job reads 50GB of order rows. A filter over those rows touches 50GB once, in the tasks that read them, and produces nothing on disk. An orderBy("order_date") over the same rows writes about 50GB of shuffle files, then reads about 50GB over the network, so the rows are handled 3 times. A dropDuplicates on the full row does the same when nearly every row is unique, because nearly every row survives to the shuffle. Sort the table and then deduplicate it and the rows cross twice: 100GB written, 100GB read, before the job has produced its first row of output.
When the aggregate is sum or count, Spark always adds up each task's rows per key before sending, and avg gets the same treatment. So a groupBy moves at most a single row per key per partition. The Shuffles: Intermediate lesson covers how.
The wait at the boundary
Cost in bytes is half the story. The readers cannot begin until the slowest writer finishes, because that writer may hold rows for any of them. Picture 4 writing tasks where 3 finish in 1 minute and the fourth, holding a partition 10 times larger, takes 10 minutes. The 3 quick tasks finish, their output sits on disk, their slots go idle, and every reader starts at minute 10. The cluster did 13 task-minutes of work across a 10-minute window in which most slots were doing nothing.
| Task | Rows | Finishes at |
|---|---|---|
| Writer 0 | 2 million | minute 1 |
| Writer 1 | 2 million | minute 1 |
| Writer 2 | 2 million | minute 1 |
| Writer 3 | 20 million | minute 10 |
| Every reader | starts | minute 10 |
A narrow chain has no such wall. Each task finishes and its result is final, so a slow partition delays only itself while the other slots move on to the next partitions. The moment you add a wide call, the slowest partition anywhere in the input sets the pace for everything after it.
Shrink the rows before they cross
Since a shuffle charges per row, the cheapest shuffle is the one with the fewest rows. A filter placed before a wide call drops rows in the task that read them, at CPU cost only. With 1,000,000 orders of which 250,000 are Completed, filtering before an orderBy means 250,000 rows cross instead of 1,000,000. Before a groupBy, it cuts the rows every writer hashes and adds up before sending. Any condition you can test on a raw row belongs before the wide call, and a condition on the aggregated result, such as a total above 1000, can only come after.
- Testable on the raw row
- status == "Completed"
- profit > 0
- region in a short list
- Put it before the groupBy: rows drop out before they are written
- Testable only after grouping
- sum(profit) > 1000
- count > 5
- avg(profit) between 2 values
- Has to follow the agg: the shuffle has already run
The program below groups 400 million click rows by user_id and runs slowly. Which change reduces the bytes the shuffle moves?
(clicks
.groupBy("user_id")
.count()
.collect())- Replace collect() with show(), so Spark moves only the first 20 rows through the shuffle to the driver
- Insert .filter(F.col("count") > 10) after .count(), so fewer rows reach the driver at the end of the job
- Insert .filter(F.col("event_type") == "purchase") before the groupBy, so writers count and send purchase clicks only
- Insert .select("user_id") after .count(), so the shuffle carries a narrower row with fewer columns in it
Run the program, then move the status filter after the agg at the end of the chain and run it again. It fails as soon as the line defining totals runs, before any action reads data, with an error that status cannot be resolved, since grouping by region left no status column. A condition on a raw column can only go before the shuffle, which is also where it costs least.
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.