Point at the Wide Op
The second half depends on pointing at the wide operation, and that comes down to a single test. Ask of each operation: can a task produce its output partition from 1 input partition alone? If yes, the operation is narrow, and Spark runs it inside the current stage. If the output for a key needs rows from every input partition, the operation is wide, Spark inserts an Exchange, and a new stage begins after it. The test is about the key rather than the operator's name. A grouped sum needs every row of a category in one place. A sort needs to know where every value falls in the global order. A join needs both rows with the same key in the same task. A filter needs only the row in front of it.
| Operation | Narrow or wide | Why |
|---|---|---|
| filter, where | Narrow | each row decides for itself |
| select, withColumn | Narrow | computed from the row alone |
| union | Narrow | partitions are concatenated, nothing moves |
| groupBy().agg() | Wide | every row of a key must meet in a single task |
| join between 2 large tables | Wide, both sides | matching keys must meet in a single task |
| join to a table under 10 MB | Narrow in effect | the small table is copied to every executor, no shuffle |
| orderBy, sort | Wide | a range Exchange places each row by its value |
| distinct, dropDuplicates | Wide | duplicates of a key must meet in a single task |
| repartition(col) | Wide | its purpose is the Exchange; a later groupBy on the same column reuses it |
Watch the example move
Below, the query from the last section runs on the 7 orders read as 2 partitions, with 2 shuffle partitions so the board stays readable. The filter runs inside the scan's tasks: no row leaves its partition to lose order 3. Each task then pre-sums its own rows per category before anything moves, so what crosses the first Exchange is a row per category per task rather than every order. After the final sums exist, a second Exchange places them by revenue range so the sort can finish, and the driver reads the sorted partitions in order.
(orders.filter(F.col("status") == "completed")
.groupBy("category").agg(F.sum("amount").alias("revenue"))
.orderBy(F.col("revenue").desc())
.collect())| order_id | category | status | amount |
|---|---|---|---|
| 1 | Books | completed | 40 |
| 2 | Games | completed | 25 |
| 3 | Books | returned | 60 |
| 4 | Music | completed | 35 |
| 5 | Games | completed | 50 |
| 6 | Books | completed | 20 |
| 7 | Music | completed | 55 |
Read the board as 3 stages. In stage 1, each task scans its partition with the filter applied and pre-sums by category before writing shuffle files hashed by category. Stage 2 finishes each category's sum from those files and writes shuffle files split by revenue range. Stage 3 sorts within each range and hands rows to collect(). Count 2 Exchanges and 3 stages, and notice what each Exchange moved: the first carried partial sums, at most 1 row per category from each task, and the second carried 3 finished rows. On the real table the proportions hold: the first Exchange moves at most categories times tasks rows, and the second moves a row per category.
| Stage | Runs | Ends with |
|---|---|---|
| 1 | scan orders, filter completed, partial sum per category | hash Exchange on category |
| 2 | final sum per category | range Exchange on revenue |
| 3 | sort by revenue descending | rows returned to the driver |
Say it the way the plan says it
The filter is narrow and runs with the scan. The groupBy on category is the first wide operation: an exchange hashed by category, 200 reducers by default. The orderBy on revenue is the second, a range exchange. That makes 2 shuffles and 3 stages, and the driver receives a row per category.
Interviewers like to add 2 cases that change the count. A join to a small table under 10 MB is broadcast: Spark ships the small table to every executor and the join runs inside the big table's current stage. The plan shows a BroadcastExchange above the small side, and that node copies the small table to the executors: no row of the big table moves and its stage doesn't break. Only Exchange hashpartitioning and Exchange rangepartitioning count as shuffles. And a groupBy on a column the data was already repartitioned by reuses that partitioning: Spark sees the rows for each key already together and skips the second Exchange. If the interviewer adds a join or a repartition to your query, apply the test again rather than adding a stage by reflex.
Where to look if it were slow
The tail often ends with 'and what would you check?', and the answer at this depth is a named tab and a named number. The SQL tab shows the plan as a tree with each Exchange drawn as a node, so you can confirm the count you gave. The Stages tab lists each stage with its Shuffle Write and Shuffle Read in bytes, which tells you how much each Exchange moved. Inside a stage, the task table shows duration per task: when the longest task is many times the median, one key holds many more rows than the others, and that is the stage to look at. For this query, you would expect stage 1 to write the most and stage 3 to be near empty.
About this lesson
This section is part of the SQL at Scale 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.