Which Line Costs the Most
The last turn of the tail is a ranking: of the lines in this transform, which costs the most? The answer at this depth comes from a single measure, rows crossing an exchange. The operator's name tells you whether rows cross at all. How many cross depends on where the operation sits in the chain and what sits before it. A sort of 50 aggregated rows and a sort of 200 million order rows are the same orderBy, and one of them is the costliest line in the job while the other is free.
Rank the running example
Put the first query on a table of 1 billion orders, of which 1 in 5 is completed, with 50 categories. The scan reads the table and is narrow. The filter passes 200 million rows and is narrow. The groupBy exchange carries partial sums rather than orders: with 2,000 scan tasks and 50 categories, at most 100,000 rows cross the network. The orderBy exchange carries 50 rows. The costliest line by rows crossing an exchange is the groupBy, and it's cheap in absolute terms. The costliest line in wall clock is the scan plus filter, which touches every byte, and that cost belongs to the table and arrives with every query over it.
| Line | Rows in | Rows crossing an exchange | Rank by shuffle cost |
|---|---|---|---|
| scan orders | 1,000,000,000 | 0 | narrow |
| filter completed | 1,000,000,000 | 0 | narrow |
| groupBy category | 200,000,000 | up to 100,000 partial sums | 1 |
| orderBy revenue | 50 | 50 | 2 |
| collect | 50 | 50, to the driver | outside the stages |
Now remove the aggregate. Ask for every completed order sorted by amount, and the orderBy is the only wide operation, with 200 million rows crossing its range exchange and 200 million rows sorted across 200 reducers. It's the same operator in a different place in the chain, and it goes from near free to the whole cost of the job.
Add a join
The interviewer's favourite extension adds a second table. Here orders gain a customer_id and a 3 row customers table holds each customer's region, and the question becomes completed revenue per region. The customers table is tiny, so Spark broadcasts it, and the program below says so with F.broadcast to make the choice visible on the board. Watch the join run inside the scan's stage: no row of orders moves for it.
- orders: order_id BIGINT PK, customer_id BIGINT FK, category VARCHAR, status VARCHAR, amount INTEGER
- customers: customer_id BIGINT PK, region VARCHAR
- orders.customer_id references customers.customer_id (N:1)
SELECT
c.region,
SUM(o.amount) AS revenue
FROM orders o
JOIN customers c
ON o.customer_id = c.customer_id
WHERE o.status = 'completed'
GROUP BY c.region(orders.filter(F.col("status") == "completed")
.join(F.broadcast(customers), "customer_id")
.groupBy("region").agg(F.sum("amount").alias("revenue"))
.collect())| order_id | customer_id | category | status | amount |
|---|---|---|---|---|
| 1 | 1 | Books | completed | 40 |
| 2 | 2 | Games | completed | 25 |
| 3 | 3 | Books | returned | 60 |
| 4 | 1 | Music | completed | 35 |
| 5 | 2 | Games | completed | 50 |
| 6 | 3 | Books | completed | 20 |
| 7 | 1 | Music | completed | 55 |
| customer_id | region |
|---|---|
| 1 | West |
| 2 | East |
| 3 | West |
Count it: 2 stages separated by 1 shuffle exchange hashed by region; the plan's BroadcastExchange is the copy of customers. The join cost a copy of the 3 row table to each executor and nothing else. West collects 150 from 4 orders, numbers 1, 4, 6 and 7, while East collects 75 from orders 2 and 5, and the driver receives 2 rows. In the ranking, the groupBy on region is the only wide line, and the join sits with the filter as narrow. Had the program left out the hint, Spark would have reached the same plan on its own, because customers is under the 10 MB threshold.
The same join when both sides are large
Replace the 3 row customers table with 500 million customer rows and the broadcast is off the table. Spark exchanges both sides on customer_id: 200 million completed orders and 500 million customer rows both go out hashed by customer. Then 200 reducers sort and merge their keys, and a third exchange hashes the joined rows by region for the sum. The join is now the costliest line by a wide margin: 700 million rows cross for it, against the partial sums of a row per region per task that cross for the aggregate. The frame changes to 3 exchanges and 4 stages, and the Stages tab would show the 2 stages feeding the join with the largest Shuffle Write.
| Line | Small customers (broadcast) | Large customers (sort-merge) |
|---|---|---|
| filter completed | narrow | narrow |
| join on customer_id | narrow in effect, 3 rows copied to each executor | both sides exchanged, 700,000,000 rows cross |
| groupBy region | a hash exchange carrying partial sums | a hash exchange carrying partial sums |
| shuffles, stages | 1, 2 | 3, 4 |
| costliest line | groupBy region | join on customer_id |
Reading the answer off the Spark UI
The UI gives the same ranking as numbers. In the Stages tab, the stage with the largest Shuffle Write is the stage that ends in the costliest exchange, and the operation that caused it is the line you name. For the broadcast version there is a single such stage and its write is small. For the sort-merge version, 2 stages write hundreds of millions of rows each, and the join is the answer. Name the operation and its key for the interviewer, then the number you would read to confirm it, and the ranking question is finished.
- Rank wide lines by rows crossing their exchange, read from Shuffle Write
- A pre-aggregated exchange carries keys times tasks rows at most
- A sort carries every row that reaches it
- A broadcast join carries the small table once per executor and no row of the large one
- A sort-merge join carries both sides in full
The capstone problems after this lesson hand you transforms you haven't seen, some with a join, some with a distinct, some with a repartition before the groupBy. For each, write the logic, say the 4 sentences, and then rank the wide lines by what reaches them.
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.