BeginnerSpark · 14 min · 3 challenges

SQL at Scale

The candidate is 3 rounds into a data engineering loop at Netflix, and the interviewer shares 1 table, playback_events, with the 4 columns profile_id, title_id, device_type and seconds_watched. The ask sounds like any SQL screen: total seconds watched per title for a day, highest first. The candidate writes it in a minute and the interviewer nods. Then comes the tail. 'That table is among the largest we keep, and the query runs as a Spark job. Walk me through what happens when it executes.' The candidate has written correct SQL and now has to say which line moves rows across the network, where the stages break, and what they would open first if the job ran for an hour. That second half is the question. This lesson builds the answer: the logic first, then the plan Spark runs, then the line that costs the most.

What you will be able to do

Recognize the 2-part shape of a SQL at scale question, however the follow-up is phrased.

Write correct aggregation and join logic and check it against real rows before saying a word about execution.

Point at each wide operation in a transform by its key, then count the shuffles and stages Spark runs.

Rank the lines of a transform by what they cost and name the Spark UI tab and number that would confirm it.

The Transform With a Tail

Daily Life

Interviews

Most SQL interview questions end when the query is right. This one keeps going. Once you've written the aggregate or join the interviewer asked for, the conversation turns to the machine: the table is too large for 1 node, so Spark executes the statement across a cluster, and the interviewer wants to hear what that execution looks like. The tail is the point of the exercise. A query that returns the right rows on a laptop and the same right rows across 2,000 cores is the same query, so the first half tests whether you can write it and the second half tests whether you know what Spark turns it into.

2 halves of the same question

The first half is ordinary SQL correctness: the right filter, the right grouping key, the right join condition, the right order. The second half asks you to read the same statement as a plan. Each clause becomes an operator, some operators keep rows where they are, and some move rows between partitions. The operators that move rows are where the job spends most of its time and memory, since every moved row crosses the network. An interviewer who hears you name those operators, say which column they move rows by, and count how many times rows move, has heard the answer. Everything else in this lesson is practice at saying those 3 things about any transform.

✓The SQL half checks

the filter drops the right rows
the grouping key matches the question
the join condition and join type are right
the ordering is the one asked for

✗The scale half checks

you name each operation that moves rows
you name the column it moves them by
you count the shuffles and the stages
you name what you would open in the Spark UI

The tail wears disguises

The follow-up rarely says the word shuffle. It usually arrives as a question about size or time. Sometimes it is about memory, or about what you would do if the job were slow. Each phrasing is the same request: describe the plan. Read the left column of the table below as the interviewer might say it, and the right column as what they're asking you to produce.

How the tail is phrased	What it is asking for
This table is 10 TB. Does your query still work?	Which operations move rows, and where the plan breaks into stages
What happens in the cluster when you hit run?	The stages in order, with the shuffle between each pair
Where would this be slow?	The wide operation that moves the most rows
Which part of this needs the most memory?	The stage that holds the most rows on a single task: the join or the aggregate
How many jobs and stages is this?	The stage count, shuffles plus 1; the Jobs tab can list several jobs for 1 action, such as the sampling job an orderBy runs first
What would you look at in the Spark UI?	The SQL tab for the plan, the Stages tab for shuffle sizes and task times

The model the tail assumes

Spark reads a table as partitions, runs a task per partition, and chains together every operation that can finish on a partition without seeing any other. Filters qualify, and so do column expressions and projections, so all of them run inside the same task as the scan. The chain breaks at an operation whose output for a key needs rows from every input partition: a grouped sum, a sort, a join between 2 large tables, a distinct. There Spark writes each task's rows to local disk keyed by the hash or range of the key, the next stage's tasks fetch their keys over the network, and work resumes. That write and fetch is the shuffle, and the boundary it creates is the stage boundary. A plan with 2 shuffles has 3 stages. A stage starts once every stage whose shuffle output it reads has finished, so the stages of a single chain run in order, while independent stages, such as the scans feeding the 2 sides of a join, can run at the same time. A stage finishes when its slowest task does.

200

shuffle partitions by default

10 MB

broadcast threshold by default

At this depth the question asks for correctness in both halves. The interviewer doesn't expect a tuning plan, a partition count argument or a cost estimate in dollars. They expect the query to be right and the description of the plan to be right: the right operations named as wide, the right columns named as keys, the right count of stages, and a named place in the UI to confirm it. A candidate who gets the query right and then says 'Spark would distribute it' has answered half. A candidate who gets the query right and misnames the filter as a shuffle has answered half and added an error. The sections that follow build the full second half for a running example and then a variation with a join.

TIP

Say the tail back before answering it: 'You are asking what Spark does with this query on a cluster, operation by operation.' It confirms the request and buys the 5 seconds you need to find the first wide operation.

Write the Logic First

Daily Life

Interviews

Logic comes first and execution second. An execution story about a query that returns the wrong rows is worth nothing, because the interviewer will correct the query and the story will no longer match it. The plan is also derived from the logic: the grouping column decides what the aggregate shuffles by, the join condition decides what the join shuffles by, and the ORDER BY decides whether a range shuffle appears at all. Change the query and you change the plan. So write the query, check it against a handful of rows, state the result out loud, and only then turn to the cluster.

The running example

Here is the table the rest of the lesson runs on: 7 orders, where each row's amount is tagged with a category and a status. The question is the one from the intro in a retail disguise: revenue per category, counting only completed orders, largest revenue first. Mapped back, category stands in for title_id and amount for seconds_watched. The filter on status replaces the filter on the day, so the plan built here is the answer to the Netflix tail. With 7 rows you can check the logic by hand and still follow each row as it moves between partitions in the next section.

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

	SELECT
	category,
	SUM(amount) AS revenue
	FROM orders
	WHERE status = 'completed'
	GROUP BY category
	ORDER BY revenue DESC

Check it by hand. Order 3 is a returned order for 60, so the WHERE drops it and Books falls from 120 to 60. Games collects 25 and 50 for 75. Music collects 35 and 55 for 90. The ORDER BY on the alias puts Music first, then Games, then Books. Saying those 3 numbers to the interviewer proves the filter sits before the aggregate, and it gives you a result to point at later, when you describe what the driver collects at the end of the job.

category	revenue
Music	90
Games	75
Books	60

The same logic as a DataFrame

The interviewer may ask for the DataFrame form, or you may prefer to write it. Spark turns both into the same plan, so the choice changes nothing about the tail. The DataFrame version reads top to bottom in the order the operations run, which makes it useful for the second half: each method call in the chain is an operator you will later label as narrow or wide.

	result = (orders
	.filter(F.col("status") == "completed")
	.groupBy("category")
	.agg(F.sum("amount").alias("revenue"))
	.orderBy(F.col("revenue").desc())
	.collect())

3 correctness traps the tail will expose

The traps below each produce a plausible query with a different plan, which is why they surface during the tail. A HAVING clause on status fails, because status is neither grouped nor aggregated. The usual patch is to add status to the GROUP BY. The query then runs and returns the right sums, but it groups by category and status, so the partial sums of returned orders cross the shuffle before HAVING discards them. A SUM over a column with NULL amounts returns the sum of the non-NULL values, which is right for revenue and wrong for a count of rows, so say which one you mean. And ORDER BY with ties returns those rows in whichever order the sort shuffle delivers them, so add a second sort key if the output feeds anything that compares runs.

✓Do

Filter on row attributes in WHERE, before the aggregate
State the result with real numbers before the tail
Name the grouping column out loud: it becomes the shuffle key
Add a tiebreaker to ORDER BY when order must be stable

✗Don't

Patch a HAVING on status by grouping on status: returned rows cross the shuffle
Use COUNT(amount) when you mean COUNT(*)
Start on execution with a query you haven't checked
Rewrite the query during the tail without saying the plan changed

Each clause is already an operator

Once the query is right, the plan is a reading of it. Spark's optimizer turns each clause into an operator, and the physical plan it runs names them. The table below maps this example's clauses to those operators. Of its 5 rows, 2 carry an Exchange, and those 2 are what you'll point at.

Clause	Physical operators	Moves rows between partitions?
FROM orders	FileScan	No
WHERE status = 'completed'	Filter	No
GROUP BY category, SUM(amount)	HashAggregate (partial), Exchange hashed by category, HashAggregate (final)	Yes, once
ORDER BY revenue DESC	Exchange by revenue range, Sort	Yes, once
collect()	result returned to the driver	Rows leave the executors for the driver, outside the stages

Calling explain() on the DataFrame prints these names, and each Exchange hashpartitioning or Exchange rangepartitioning node marks a stage boundary.

Point at the Wide Op

Daily Life

Interviews

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.

1(orders.filter(F.col("status") == "completed")
2   .groupBy("category").agg(F.sum("amount").alias("revenue"))
3   .orderBy(F.col("revenue").desc())
4   .collect())

orders7 rows

order_id

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.

The 3 numbers to name

▸Exchange count, read from the plan tree in the SQL tab
▸Shuffle Write per stage, read from the Stages tab
▸Longest task against median task within a stage, read from the task table

TIP

Every wide operation has a key. When you name the operation, name the column with it: 'a hash exchange on category', 'a range exchange on revenue'. The follow-up questions will be about that column.

Correct Logic, Frozen

Daily Life

Interviews

Here is the failure this question is built to catch. After writing a correct query and checking the result out loud, the candidate hears the tail and stops. What follows is one of 3 things: silence, a recital about Spark being distributed and lazy that never touches the query, or a jump to remedies such as adding executors or caching before any operation has been named. Each leaves the interviewer with the same note: knows SQL, cannot read a plan. The fix is a frame you can produce for any transform in 4 sentences, built from the test in the last section.

3 shapes of the freeze

What the candidate says	What the interviewer hears
Spark would distribute it across the cluster.	Nothing about this query: no operation, no key, no count
I would cache the table and add more executors.	A remedy for a problem that has not been located
The WHERE clause would be the expensive part on 10 TB.	A narrow operation named as the cost, which is an error
It depends on the data.	A true sentence that ends the conversation

The third row is the one to fear, because it adds an error to a correct query. A filter on 10 TB reads 10 TB, and that read is the scan's cost, which exists for every query over the table. It moves nothing between partitions. Naming it as the expensive part tells the interviewer you cannot tell a scan from a shuffle, and the rest of the tail will probe that until it is confirmed.

The 4-sentence frame

Say these 4 things, in order

▸First: the narrow operations and the stage they share with the scan.
▸Second: the first wide operation and the exchange it causes on its key.
▸Third: each further wide operation with its key, then the total count of shuffles and stages.
▸Fourth: the place in the Spark UI you would confirm it, and the number you would read there.

Here is the frame filled in for the running example, as you would say it. Read it for rhythm: every sentence names an operation and a column, and nothing in it is a remedy.

The filter on status is narrow and stays in the scan's stage. The first wide operation is the groupBy, a hash exchange on category, and each task pre-sums before it. The orderBy on revenue adds a range exchange, so the plan has 2 shuffles and 3 stages. In the Stages tab I would read Shuffle Write for stage 1, which should be small because each task sends one partial total for each category it saw, never the orders themselves.

When the frame gets a follow-up

The interviewer will push on 1 of the 4 sentences, and the push stays inside the plan you described, so the frame has already told you where you are. 'What if there are 2 billion completed orders?' pushes on the second sentence: the partial aggregate still caps each task's output at one total for each category, so the exchange stays small, and the scan is what grows. 'What if 1 category is 90% of orders?' pushes on the same exchange: the final sum for that category lands on a single reducer, and the partial sums keep it to a row per task, so the aggregate survives it, where a join on that column would send every one of those rows to a single task. 'What if we drop the ORDER BY?' pushes on the third sentence: 1 shuffle, 2 stages.

✓The frozen answer

starts with a remedy
speaks about Spark in general
names no column
cannot say how many stages

✗The plan-reading answer

starts with the first wide operation
speaks about this query
names the key of every exchange
gives a shuffle count and a stage count

One more freeze shows up in the join variant: the candidate writes a correct join and then declares every join a shuffle. Apply the size test from the last section in place of the reflex, and if you can't know the dimension's size, say that the plan depends on it and give the count for each case.

To practice the frame, take any query you have written this week and, with the plan covered, say the 4 sentences before running explain(). Then compare: every Exchange hashpartitioning or Exchange rangepartitioning in the output should match a wide operation you named, and nothing you named as narrow should appear beside one. After a week of this, the first wide operation in a new query shows up as you read it.

TIP

If you go blank, say the first sentence of the frame out loud: 'The filter is narrow.' It is always true of a filter, and saying it starts the walk through the rest of the chain.

Which Line Costs the Most

Daily Life

Interviews

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_idPKBIGINT

customer_idFKBIGINT

categoryVARCHAR

statusVARCHAR

amountINTEGER

customers

customer_idPKBIGINT

regionVARCHAR

	SELECT
	c.region,
	SUM(o.amount) AS revenue
	FROM orders AS o
	JOIN customers AS c
	ON o.customer_id = c.customer_id
	WHERE o.status = 'completed'
	GROUP BY c.region

1(orders.filter(F.col("status") == "completed")
2   .join(F.broadcast(customers), "customer_id")
3   .groupBy("region").agg(F.sum("amount").alias("revenue"))
4   .collect())

orders7 rows

order_id

customer_id

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.

❯❯❯PUTTING IT ALL TOGETHER

> A marketplace's finance team asks for yesterday's completed revenue by seller region before the 9 am review. The orders table holds 1 billion rows, about 200 million of them completed, and the sellers table holds 40,000 rows with a region each. The job runs on a Spark cluster with the default of 200 shuffle partitions.

Logic first: the WHERE on status drops uncompleted orders before the aggregate, the join on seller_id attaches the region, and the GROUP BY on region produces one row per region, which is the result to state before any talk of the cluster.

Point at the wide op: the sellers table is far under 10 MB, so Spark broadcasts it and the join adds no shuffle; the groupBy on region is the only wide operation, so the plan has 1 shuffle and 2 stages.

Correct logic, frozen: the mistakes to avoid here are declaring the join a shuffle or naming the 1 billion row filter as the expensive part; the frame says filter narrow, join broadcast, groupBy on region as the one exchange.

Which line costs the most: the scan and filter touch every byte but move nothing, and the groupBy exchange carries at most regions times tasks partial sums, so its Shuffle Write in the Stages tab is small; if sellers grew to hundreds of millions of rows, the join would become a sort-merge with both sides exchanged and would take over as the costliest line.

KEY TAKEAWAYS

A SQL at scale question has 2 halves: the correct query, then the plan Spark runs it as, and the second half is what the interviewer is grading.

Write and check the logic first, because the exchanges fall where the grouping column and join condition put them, and the ORDER BY adds one more.

The test for a wide operation is whether a key's output needs rows from every input partition: filters and projections are narrow, while grouped sums and sorts are wide, as are distincts and joins between large tables.

Count shuffles and stages from the plan: a join to a table under 10 MB is broadcast and adds no shuffle, an orderBy adds a range exchange, and a groupBy after a repartition on the same key reuses it.

Rank the lines by rows crossing an exchange: a pre-aggregated groupBy moves keys times tasks rows at most, a sort moves everything that reaches it, a sort-merge join moves both sides.

Name the Spark UI check with the operation: the SQL tab for the exchange count, the Stages tab for Shuffle Write per stage, the task table for the longest task against the median.

Write the query, then say where Spark shuffles it: the 2 halves of one interview question.

Category: Spark
Difficulty: beginner
Duration: 14 minutes
Challenges: 3 hands-on challenges

Topics covered: The Transform With a Tail, Write the Logic First, Point at the Wide Op, Correct Logic, Frozen, Which Line Costs the Most

Lesson Sections

The Transform With a Tail
Most SQL interview questions end when the query is right. This one keeps going. Once you've written the aggregate or join the interviewer asked for, the conversation turns to the machine: the table is too large for 1 node, so Spark executes the statement across a cluster, and the interviewer wants to hear what that execution looks like. The tail is the point of the exercise. A query that returns the right rows on a laptop and the same right rows across 2,000 cores is the same query, so the first
Write the Logic First
Logic comes first and execution second. An execution story about a query that returns the wrong rows is worth nothing, because the interviewer will correct the query and the story will no longer match it. The plan is also derived from the logic: the grouping column decides what the aggregate shuffles by, the join condition decides what the join shuffles by, and the ORDER BY decides whether a range shuffle appears at all. Change the query and you change the plan. So write the query, check it agai
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 ro
Correct Logic, Frozen
Here is the failure this question is built to catch. After writing a correct query and checking the result out loud, the candidate hears the tail and stops. What follows is one of 3 things: silence, a recital about Spark being distributed and lazy that never touches the query, or a jump to remedies such as adding executors or caching before any operation has been named. Each leaves the interviewer with the same note: knows SQL, cannot read a plan. The fix is a frame you can produce for any trans
Which Line Costs the Most
Rank the running example Add a join The same join when both sides are large 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 answ