Apache Spark Plan Reading Interview Questions (21, with answers)

01 / 21

Shuffle mechanicsJoin strategies (broadcast / SMJ / skew join / salting)Lazy evaluation & Catalyst

Filter orders to EU, join customers, group by tier. How many shuffles does Spark do, and which line of the code caused each one?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

The code — what does Spark do with it?

o = spark.table("orders")
c = spark.table("customers")
out = (o.filter(o.region == "EU")
        .join(c, "customer_id")
        .groupBy("tier").agg(F.sum("amount").alias("total")))

The code — predict the output before reading on

o = spark.table("orders")
c = spark.table("customers")
out = (o.filter(o.region == "EU")
        .join(c, "customer_id")
        .groupBy("tier").agg(F.sum("amount").alias("total")))

Why they ask this

The opening question of almost every Spark plan round. It is asked because the shuffle count is the single best predictor of what a job costs, and because reading it off a plan requires knowing what an Exchange is.

Say this

One. The join is a broadcast so it costs no shuffle at all, and the only Exchange in the plan is the one the groupBy needs to bring each tier together.

The reasoning

Count Exchange nodes — that is what a shuffle is in a physical plan. There is exactly one here, Exchange hashpartitioning(tier, 200), sitting between the two HashAggregate nodes. That is the classic partial-then-final aggregation: Spark aggregates within each partition first, shuffles the much smaller partial results, then combines them.

The join contributed nothing, which is the part worth saying out loud. customers is small enough to sit under the broadcast threshold, so Catalyst chose BroadcastHashJoin — the small side is collected to the driver, shipped to every executor, and the join happens inside the existing partitions. The BroadcastExchange in the tree is a broadcast, not a shuffle: no repartitioning by key, no sort, no shuffle files.

The filter also disappeared from the tree as a separate step, because Catalyst pushed it into the scan — you can see it in the scan's PushedFilters. So the answer to the follow-up question, which is always 'and how many stages', is read off the codegen markers: the *(1), *(2) prefixes number the whole-stage codegen groups, and a new one starts on each side of an Exchange.

What it actually returns 1 shuffle, run on Spark 4.2

Spark 4.2, 200 shuffle partitions, AQE off, broadcast threshold 10MB.

The physical plan1 shuffle

== Physical Plan ==
*(3) HashAggregate(keys=[tier#1], functions=[sum(amount#2)])
+- Exchange hashpartitioning(tier#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(2) HashAggregate(keys=[tier#1], functions=[partial_sum(amount#2)])
      +- *(2) Project [amount#2, tier#1]
         +- *(2) BroadcastHashJoin [customer_id#3], [customer_id#4], Inner, BuildRight, false, false
            :- *(2) Project [customer_id#3, amount#2]
            :  +- *(2) Filter ((isnotnull(region#5) AND (region#5 = EU)) AND isnotnull(customer_id#3))
            :     +- *(2) ColumnarToRow
            :        +- BatchScan parquet /tables/orders[customer_id#3, region#5, amount#2] ParquetScan DataFilters: [isnotnull(region#5), (region#5 = EU), isnotnull(customer_id#3)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(region), EqualTo(region,EU), IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,region:string,amount:int> RuntimeFilters: []
            +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=N]
               +- *(1) Filter isnotnull(customer_id#4)
                  +- *(1) ColumnarToRow
                     +- BatchScan parquet /tables/customers[customer_id#4, tier#1] ParquetScan DataFilters: [isnotnull(customer_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []

It returns

tier	total
free	112928
pro	112041

The answer most people give

"Two — one for the join and one for the group by." Only if the join were a sort-merge join. With customers under the broadcast threshold there is no Exchange on either side of it, and reading BroadcastExchange as a shuffle is the most common misreading of a Spark plan.

They’ll ask next

customers grows to 500 MB. Redraw the plan in words, and say how many shuffles it has now.

The Shuffle — Spark's Performance Killer Spark SQL & the Catalyst Optimizer

02 / 21

Shuffle mechanicsJoin strategies (broadcast / SMJ / skew join / salting)

Same join and aggregation, but autoBroadcastJoinThreshold is -1 so nothing can be broadcast. How many shuffles now, and where did the Sort nodes come from?

The setup — predict the table before reading on

spark.sql.autoBroadcastJoinThreshold-1

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

The code — what does Spark do with it?

o = spark.table("orders")
c = spark.table("customers")
out = (o.join(c, "customer_id")
        .groupBy("tier").agg(F.sum("amount").alias("total")))

The code — predict the output before reading on

o = spark.table("orders")
c = spark.table("customers")
out = (o.join(c, "customer_id")
        .groupBy("tier").agg(F.sum("amount").alias("total")))

Why they ask this

It is the paired question, and it separates people who can read a plan from people who have memorised one. The Sort nodes are the giveaway that this is a sort-merge join.

Say this

Three: one Exchange per side of the join, plus the one for the aggregation. The Sorts are there because a sort-merge join requires both sides sorted by the join key inside each partition.

The reasoning

Disabling broadcast forces SortMergeJoin, and that costs a lot more than the strategy name suggests. Both sides have to be hash-partitioned on customer_id so matching keys land on the same executor — that is two Exchanges — and then each partition has to be sorted by the key so the merge can walk both sides in step. Add the aggregation's Exchange and the total is three.

The codegen markers tell you how the work splits into stages: the plan runs from *(1) up to *(6), and each Exchange is a stage boundary because a stage cannot span a shuffle. That is the mechanical link between the plan and the Spark UI — the stages you see there are the segments between Exchange nodes.

Notice also Filter isnotnull(customer_id) appearing on both sides. Catalyst inserts it because an inner join on a key can never match a null, so discarding nulls early is free and shrinks the shuffle. It is a small thing and it is a good signal in an interview — it shows you are reading the plan rather than recognising it.

What it actually returns 3 shuffles, run on Spark 4.2

Identical code to the previous question. Only the broadcast threshold changed.

The physical plan3 shuffles

== Physical Plan ==
*(6) HashAggregate(keys=[tier#1], functions=[sum(amount#2)])
+- Exchange hashpartitioning(tier#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(5) HashAggregate(keys=[tier#1], functions=[partial_sum(amount#2)])
      +- *(5) Project [amount#2, tier#1]
         +- *(5) SortMergeJoin [customer_id#3], [customer_id#4], Inner
            :- *(2) Sort [customer_id#3 ASC NULLS FIRST], false, 0
            :  +- Exchange hashpartitioning(customer_id#3, 200), ENSURE_REQUIREMENTS, [plan_id=N]
            :     +- *(1) Filter isnotnull(customer_id#3)
            :        +- *(1) ColumnarToRow
            :           +- BatchScan parquet /tables/orders[customer_id#3, amount#2] ParquetScan DataFilters: [isnotnull(customer_id#3)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,amount:int> RuntimeFilters: []
            +- *(4) Sort [customer_id#4 ASC NULLS FIRST], false, 0
               +- Exchange hashpartitioning(customer_id#4, 200), ENSURE_REQUIREMENTS, [plan_id=N]
                  +- *(3) Filter isnotnull(customer_id#4)
                     +- *(3) ColumnarToRow
                        +- BatchScan parquet /tables/customers[customer_id#4, tier#1] ParquetScan DataFilters: [isnotnull(customer_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []

It returns

tier	total
free	337010
pro	337220

The answer most people give

"Two — one shuffle per side of the join." That misses the aggregation's own Exchange, which is the one that is always there. Counting join shuffles and forgetting the groupBy is the most common way to undercount.

They’ll ask next

Which of these three shuffles would you attack first, and how?

Joins in Spark The Shuffle — Spark's Performance Killer

03 / 21

Shuffle mechanicsLazy evaluation & Catalyst

This aggregates orders per customer and then joins the customer dimension. Is the shuffle count different from joining first and aggregating after, and why would you write it this way?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

The code — what does Spark do with it?

o = spark.table("orders")
c = spark.table("customers")
per_customer = o.groupBy("customer_id").agg(F.sum("amount").alias("spend"))
out = per_customer.join(c, "customer_id").select("customer_id", "tier", "spend")

The code — predict the output before reading on

o = spark.table("orders")
c = spark.table("customers")
per_customer = o.groupBy("customer_id").agg(F.sum("amount").alias("spend"))
out = per_customer.join(c, "customer_id").select("customer_id", "tier", "spend")

Why they ask this

It is a question about ordering operations to shrink what gets shuffled — the single most effective tuning move there is, and one people apply by habit without being able to justify.

Say this

Still one shuffle, but a far smaller one: the aggregation happens on orders alone, so what crosses the network is one row per customer rather than every order widened with customer columns.

The reasoning

The Exchange count is the same, and that is the trap in the question — shuffle *count* is a poor proxy for shuffle *cost*. What matters is how many bytes cross the network, and aggregating before the join means the shuffle carries 400 pre-aggregated rows instead of 2,000 order rows.

The join itself is free here for the same reason as before: customers is under the broadcast threshold, so the plan shows BroadcastHashJoin with no Exchange attached to it. The one Exchange in the tree belongs to the groupBy on customer_id.

The general principle is to reduce before you widen — filter, aggregate and project away columns as early as the semantics allow, so that every operation downstream moves less data. Catalyst will do a lot of this for you: it pushes filters and prunes columns without being asked. It will not reorder an aggregation across a join, because that changes the semantics unless it can prove otherwise, so this one is yours to write.

What it actually returns 1 shuffle, run on Spark 4.2

One Exchange, carrying 400 rows instead of 2,000.

The physical plan1 shuffle

== Physical Plan ==
*(3) Project [customer_id#1, tier#2, spend#3L]
+- *(3) BroadcastHashJoin [customer_id#1], [customer_id#4], Inner, BuildRight, false, false
   :- *(3) HashAggregate(keys=[customer_id#1], functions=[sum(amount#5)])
   :  +- Exchange hashpartitioning(customer_id#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   :     +- *(1) HashAggregate(keys=[customer_id#1], functions=[partial_sum(amount#5)])
   :        +- *(1) Filter isnotnull(customer_id#1)
   :           +- *(1) ColumnarToRow
   :              +- BatchScan parquet /tables/orders[customer_id#1, amount#5] ParquetScan DataFilters: [isnotnull(customer_id#1)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,amount:int> RuntimeFilters: []
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=N]
      +- *(2) Filter isnotnull(customer_id#4)
         +- *(2) ColumnarToRow
            +- BatchScan parquet /tables/customers[customer_id#4, tier#2] ParquetScan DataFilters: [isnotnull(customer_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []

It returns

customer_id	tier	spend
0	free	865
1	pro	900
2	free	935
3	pro	970
4	free	1005
5	pro	1040
6	free	1075
7	pro	1110
8	free	1145
9	pro	1180
10	free	1215
11	pro	1250

The answer most people give

"Joining first is the same because Catalyst will reorder it." Catalyst reorders filters and projections freely; it will not move an aggregation through a join in the general case, because doing so is only valid under conditions it usually cannot prove.

They’ll ask next

When would aggregating first be wrong — what would the join have to look like?

Aggregations & Grouping Spark SQL & the Catalyst Optimizer

04 / 21

Partitioning, repartition vs coalesceShuffle mechanics

Someone added .repartition(50, "customer_id") before a groupBy on region, believing it would speed the aggregation up. What did it actually do to the plan?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

The code — what does Spark do with it?

out = (spark.table("orders")
        .groupBy("region").agg(F.sum("amount").alias("total")))

The job as written — it works, and it is slow

out = (spark.table("orders")
        .repartition(50, "customer_id")
        .groupBy("region").agg(F.sum("amount").alias("total")))

The physical plan2 shuffles

== Physical Plan ==
*(3) HashAggregate(keys=[region#1], functions=[sum(amount#2)])
+- Exchange hashpartitioning(region#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(2) HashAggregate(keys=[region#1], functions=[partial_sum(amount#2)])
      +- *(2) Project [region#1, amount#2]
         +- Exchange hashpartitioning(customer_id#3, 50), REPARTITION_BY_NUM, [plan_id=N]
            +- *(1) ColumnarToRow
               +- BatchScan parquet /tables/orders[customer_id#3, region#1, amount#2] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,region:string,amount:int> RuntimeFilters: []

It returns

region	total
APAC	224376
EU	224969
US	224885

The code — predict the output before reading on

out = (spark.table("orders")
        .groupBy("region").agg(F.sum("amount").alias("total")))

Why they ask this

repartition is the most misapplied call in PySpark. It is reached for as a general-purpose 'make it faster', and the plan shows exactly what it costs.

Say this

It added a whole extra shuffle. The aggregation needs the data partitioned by region, and partitioning by customer_id does not satisfy that — so Spark shuffles once for the repartition and again for the groupBy.

The reasoning

A shuffle is only avoidable when the data already satisfies the distribution the next operator requires. The groupBy on region requires hashpartitioning(region); repartitioning by customer_id gives hashpartitioning(customer_id), which is not that, so Spark has to do it again. The plan below shows both Exchanges — the rewrite has one, the version as written has two.

The rewrite here is simply deleting the line. That is the honest answer to a lot of repartition calls: the aggregation was always going to shuffle, and adding a shuffle in front of it does not help. Both versions return identical rows, which the harness asserts, so this is a pure cost saving.

Repartitioning is worth it when the distribution it produces is *reused* — before several aggregations or joins on the same key, or when you deliberately want a different number of output files. The next question shows the case where the same call is free rather than wasteful, which is the distinction to be able to draw.

What it actually returns 1 shuffle, run on Spark 4.2

Same rows, one fewer Exchange — asserted by the harness.

The physical plan1 shuffle

== Physical Plan ==
*(2) HashAggregate(keys=[region#1], functions=[sum(amount#2)])
+- Exchange hashpartitioning(region#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(1) HashAggregate(keys=[region#1], functions=[partial_sum(amount#2)])
      +- *(1) ColumnarToRow
         +- BatchScan parquet /tables/orders[region#1, amount#2] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<region:string,amount:int> RuntimeFilters: []

It returns

region	total
APAC	224376
EU	224969
US	224885

The answer most people give

"repartition improves parallelism, so it helps." It sets the partition count, and the aggregation was going to set it anyway via spark.sql.shuffle.partitions. What you have added is a full network shuffle for no change in the result.

They’ll ask next

When is repartitioning before an aggregation actually worth it?

Partitioning & Parallelism The Shuffle — Spark's Performance Killer

05 / 21

Partitioning, repartition vs coalesceShuffle mechanics

Same code, but repartitioned on region — the column the groupBy uses. How many Exchanges are in the plan now?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

The code — what does Spark do with it?

out = (spark.table("orders")
        .repartition(50, "region")
        .groupBy("region").agg(F.sum("amount").alias("total")))

The code — predict the output before reading on

out = (spark.table("orders")
        .repartition(50, "region")
        .groupBy("region").agg(F.sum("amount").alias("total")))

Why they ask this

The paired question, and the one that shows the rule rather than the anecdote. Candidates who learned 'repartition is bad' get this wrong in the other direction.

Say this

Still one. The repartition produces exactly the distribution the aggregation requires, so Spark uses it and adds nothing — the Exchange in the plan is the repartition, doing the aggregation's work for it.

The reasoning

Catalyst tracks the output partitioning of every operator and compares it against what the next one requires. repartition(50, "region") produces hashpartitioning(region, 50); the aggregation requires hashpartitioning(region, anything); the requirement is satisfied, so no second Exchange is inserted.

That is the actual rule, and it is worth stating in those terms: a shuffle is inserted when the required distribution is not already satisfied. Everything else about repartition, coalesce and bucketing follows from it — bucketing is exactly this trick made durable, storing the data pre-partitioned on disk so that even the first shuffle disappears.

The one thing that did change is the partition count: 50 instead of the 200 that spark.sql.shuffle.partitions would have given. That is a real lever — 200 partitions over a small result means 200 mostly-empty tasks, each with scheduling overhead — and it is the legitimate reason to write this call.

What it actually returns 1 shuffle, run on Spark 4.2

The Exchange is the repartition. The aggregation needed no second one.

The physical plan1 shuffle

== Physical Plan ==
*(2) HashAggregate(keys=[region#1], functions=[sum(amount#2)])
+- *(2) HashAggregate(keys=[region#1], functions=[partial_sum(amount#2)])
   +- Exchange hashpartitioning(region#1, 50), REPARTITION_BY_NUM, [plan_id=N]
      +- *(1) ColumnarToRow
         +- BatchScan parquet /tables/orders[region#1, amount#2] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<region:string,amount:int> RuntimeFilters: []

It returns

region	total
APAC	224376
EU	224969
US	224885

The answer most people give

"Two, because repartition always shuffles." repartition always *requests* a distribution; whether a second shuffle follows depends on whether the next operator needs a different one. Here it does not.

They’ll ask next

How would bucketing the table on region change this plan?

Partitioning & Parallelism Broadcast Joins, Shared Variables & Bucketing

06 / 21

Join strategies (broadcast / SMJ / skew join / salting)Lazy evaluation & Catalyst

This join produced a BroadcastHashJoin with BuildRight and zero shuffles. What did the optimizer know that made that decision, and what would take it away?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

The code — what does Spark do with it?

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .select("order_id", "tier", "amount"))

The code — predict the output before reading on

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .select("order_id", "tier", "amount"))

Why they ask this

Join strategy selection is the most consequential thing Catalyst does, and it is driven by a statistic most candidates cannot say the source of.

Say this

It knew customers was smaller than autoBroadcastJoinThreshold — from the Parquet file sizes, since there are no table statistics here. BuildRight says it chose the right side to broadcast.

The reasoning

The decision is a size comparison. Catalyst estimates each side's size and, if one is below spark.sql.autoBroadcastJoinThreshold (10 MB by default), it broadcasts that side rather than shuffling both. BuildRight in the plan names which side was chosen — the build side is the one collected and shipped, the stream side is the one scanned in place.

Where the estimate comes from matters, because it is what goes wrong. With no collected statistics, Spark falls back to the size of the files on disk — which is *compressed* Parquet, and can be several times smaller than the in-memory representation. A table that looks like 8 MB on disk can be well over the threshold once it is deserialised into a hash table on the driver, which is how a broadcast join turns into a driver OOM.

Three things take the broadcast away: the side growing past the threshold, the threshold being set to -1, or the join being one broadcast cannot serve — a full outer join, or a left join where the *left* side is the small one, since you can only broadcast the side you are not preserving. Being able to name that last case is a strong signal.

What it actually returns 0 shuffles, run on Spark 4.2

Zero Exchange nodes. BroadcastExchange is not one.

The physical plan0 shuffles

== Physical Plan ==
*(2) Project [order_id#1, tier#2, amount#3]
+- *(2) BroadcastHashJoin [customer_id#4], [customer_id#5], Inner, BuildRight, false, false
   :- *(2) Filter isnotnull(customer_id#4)
   :  +- *(2) ColumnarToRow
   :     +- BatchScan parquet /tables/orders[order_id#1, customer_id#4, amount#3] ParquetScan DataFilters: [isnotnull(customer_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<order_id:int,customer_id:int,amount:int> RuntimeFilters: []
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=N]
      +- *(1) Filter isnotnull(customer_id#5)
         +- *(1) ColumnarToRow
            +- BatchScan parquet /tables/customers[customer_id#5, tier#2] ParquetScan DataFilters: [isnotnull(customer_id#5)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []

It returns

order_id	tier	amount
0	free	5
1	pro	12
2	free	19
3	pro	26
4	free	33
5	pro	40
6	free	47
7	pro	54
8	free	61
9	pro	68
10	free	75
11	pro	82

The answer most people give

"It broadcast because I wrote the small table on the right." Side order in your code does not decide it — Catalyst picks the build side from the estimates, and will broadcast the left side just as happily if that is the small one.

They’ll ask next

Why can a full outer join never be a broadcast hash join?

Broadcast Joins, Shared Variables & Bucketing Joins in Spark

07 / 21

Join strategies (broadcast / SMJ / skew join / salting)AQE

Identical code to the previous question, with autoBroadcastJoinThreshold set to 1 byte. What is the plan now, and what does that tell you about tuning by config?

The setup — predict the table before reading on

spark.sql.autoBroadcastJoinThreshold1

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

The code — what does Spark do with it?

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .select("order_id", "tier", "amount"))

The code — predict the output before reading on

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .select("order_id", "tier", "amount"))

Why they ask this

It demonstrates that a plan is a function of configuration as much as of code — which is why a plan quoted without its settings is not evidence of anything.

Say this

SortMergeJoin with two Exchanges and two Sorts, from exactly the same code. The plan is a property of code plus configuration plus statistics, never of code alone.

The reasoning

Nothing in the query changed. One session setting moved and the physical plan went from zero shuffles to two, plus the sorts. That is the single most useful thing to understand about .explain(): it shows you a decision, and the decision has inputs you can see and change.

It is also why the configuration is printed beside every plan in this bank. A colleague who shows you a plan without telling you their shuffle partitions, their AQE setting and their broadcast threshold has shown you something you cannot reason about — and the same is true of a plan in a blog post.

In production the direction people usually move this setting is up, to force broadcasts of dimensions in the tens of megabytes. That works and it has a limit: the build side is collected to the driver first, so raising the threshold raises driver memory pressure, and every executor holds its own copy. Raising it to several hundred megabytes on a cluster with many executors is how you turn a join problem into a memory problem.

What it actually returns 2 shuffles, run on Spark 4.2

autoBroadcastJoinThreshold = 1. Same query, different plan.

The physical plan2 shuffles

== Physical Plan ==
*(5) Project [order_id#1, tier#2, amount#3]
+- *(5) SortMergeJoin [customer_id#4], [customer_id#5], Inner
   :- *(2) Sort [customer_id#4 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(customer_id#4, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   :     +- *(1) Filter isnotnull(customer_id#4)
   :        +- *(1) ColumnarToRow
   :           +- BatchScan parquet /tables/orders[order_id#1, customer_id#4, amount#3] ParquetScan DataFilters: [isnotnull(customer_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<order_id:int,customer_id:int,amount:int> RuntimeFilters: []
   +- *(4) Sort [customer_id#5 ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(customer_id#5, 200), ENSURE_REQUIREMENTS, [plan_id=N]
         +- *(3) Filter isnotnull(customer_id#5)
            +- *(3) ColumnarToRow
               +- BatchScan parquet /tables/customers[customer_id#5, tier#2] ParquetScan DataFilters: [isnotnull(customer_id#5)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []

It returns

order_id	tier	amount
0	free	5
1	pro	12
2	free	19
3	pro	26
4	free	33
5	pro	40
6	free	47
7	pro	54
8	free	61
9	pro	68
10	free	75
11	pro	82

The answer most people give

"The plan is determined by the query." It is determined by the query, the configuration and the statistics. Two of those three are invisible in the code, which is why reproducing someone else's plan needs their settings.

They’ll ask next

Your dimension is 40 MB. Do you raise the threshold or hint the broadcast, and what is the difference?

Broadcast Joins, Shared Variables & Bucketing Performance Tuning & Debugging with the Spark UI

08 / 21

Join strategies (broadcast / SMJ / skew join / salting)AQE

Broadcast is disabled by the threshold, but F.broadcast() is applied to the small side. Which wins, and what have you taken responsibility for by writing that?

The setup — predict the table before reading on

spark.sql.autoBroadcastJoinThreshold-1

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

The code — what does Spark do with it?

out = (spark.table("orders")
        .join(F.broadcast(spark.table("customers")), "customer_id")
        .select("order_id", "tier", "amount"))

The code — predict the output before reading on

out = (spark.table("orders")
        .join(F.broadcast(spark.table("customers")), "customer_id")
        .select("order_id", "tier", "amount"))

Why they ask this

Hints are how tuning actually gets done, and the interviewer wants to know whether you understand what you have overridden — because a hint silences the safety check.

Say this

The hint wins: the plan is a BroadcastHashJoin with zero shuffles even with the threshold at -1. What you have taken on is the size judgement Catalyst was making for you, permanently, including as the table grows.

The reasoning

F.broadcast(df) is an instruction rather than a suggestion — it overrides the threshold entirely, which is exactly why it is useful when the estimate is wrong. Estimates are wrong quite often: after a chain of filters and joins, Catalyst's idea of a DataFrame's size can be far off, and a hint is how you tell it what you know.

The cost is that you have hard-coded a judgement about size into code that will outlive the data it was written against. The table is 12 MB today; in a year it is 800 MB and the hint is still there, collecting all of it to the driver and shipping it to every executor. The failure is a driver OOM or a broadcast timeout, and it arrives long after anyone remembers writing the hint.

So use it deliberately and leave a note: what the size was, and what should happen if it grows. The alternative worth knowing is that AQE can convert a sort-merge join into a broadcast at runtime when the actual shuffled size turns out to be small — which is the same win without the hard-coded assumption, and it is one of the strongest arguments for leaving AQE on.

What it actually returns 0 shuffles, run on Spark 4.2

Threshold at -1, and the hint broadcasts anyway.

The physical plan0 shuffles

== Physical Plan ==
*(2) Project [order_id#1, tier#2, amount#3]
+- *(2) BroadcastHashJoin [customer_id#4], [customer_id#5], Inner, BuildRight, false, false
   :- *(2) Filter isnotnull(customer_id#4)
   :  +- *(2) ColumnarToRow
   :     +- BatchScan parquet /tables/orders[order_id#1, customer_id#4, amount#3] ParquetScan DataFilters: [isnotnull(customer_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<order_id:int,customer_id:int,amount:int> RuntimeFilters: []
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=N]
      +- *(1) Filter isnotnull(customer_id#5)
         +- *(1) ColumnarToRow
            +- BatchScan parquet /tables/customers[customer_id#5, tier#2] ParquetScan DataFilters: [isnotnull(customer_id#5)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []

It returns

order_id	tier	amount
0	free	5
1	pro	12
2	free	19
3	pro	26
4	free	33
5	pro	40
6	free	47
7	pro	54
8	free	61
9	pro	68
10	free	75
11	pro	82

The answer most people give

"The threshold wins because it is a hard limit." It is a threshold for the automatic decision, not a cap on what can be broadcast. An explicit hint bypasses it, which is the whole point of the hint and also its danger.

They’ll ask next

How does AQE achieve the same thing without a hint, and when can it not?

Broadcast Joins, Shared Variables & Bucketing Adaptive Query Execution & Cost-Based Optimization

09 / 21

Join strategies (broadcast / SMJ / skew join / salting)Shuffle mechanics

orders joined to customers joined to products, then grouped by tier and category. Two joins and an aggregation — why is there only one Exchange?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

products(product_id int, category string, list_price int)

200 products in 8 categories.

products

product_id	category	list_price
0	cat_0	10
1	cat_1	13
2	cat_2	16
3	cat_3	19

The code — what does Spark do with it?

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .join(spark.table("products"), "product_id")
        .groupBy("tier", "category").agg(F.sum("amount").alias("total")))

The code — predict the output before reading on

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .join(spark.table("products"), "product_id")
        .groupBy("tier", "category").agg(F.sum("amount").alias("total")))

Why they ask this

It checks whether the candidate generalises the broadcast rule rather than counting joins, and whether they can read a plan with more than one join in it.

Say this

Both dimensions are small enough to broadcast, so neither join shuffles. The single Exchange belongs to the groupBy — which is the shape most star-schema jobs have when the dimensions are sized properly.

The reasoning

This is the plan you want a star-schema job to have: the large fact table is scanned once, each dimension is broadcast into the same set of partitions, and the only network movement is the aggregation's. Both joins are BroadcastHashJoin and neither contributes an Exchange.

It generalises, and that is the point of the question. The rule is per join, not per query: every join whose build side fits under the threshold is free of shuffle. A job joining six dimensions to one fact can still have exactly one Exchange, which is why a well-modelled star schema performs so much better than the same data normalised into pieces that all need shuffling.

The failure mode to watch is the one this design makes easy to miss: each broadcast is held in memory on every executor simultaneously. Six dimensions at 8 MB each is 48 MB per executor before any of your data is loaded — usually fine, and worth knowing about when executors are small and the dimension count is high.

What it actually returns 1 shuffle, run on Spark 4.2

Two joins, zero join shuffles. The Exchange is the aggregation's.

The physical plan1 shuffle

== Physical Plan ==
*(4) HashAggregate(keys=[tier#1, category#2], functions=[sum(amount#3)])
+- Exchange hashpartitioning(tier#1, category#2, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(3) HashAggregate(keys=[tier#1, category#2], functions=[partial_sum(amount#3)])
      +- *(3) Project [amount#3, tier#1, category#2]
         +- *(3) BroadcastHashJoin [product_id#4], [product_id#5], Inner, BuildRight, false, false
            :- *(3) Project [product_id#4, amount#3, tier#1]
            :  +- *(3) BroadcastHashJoin [customer_id#6], [customer_id#7], Inner, BuildRight, false, false
            :     :- *(3) Filter (isnotnull(customer_id#6) AND isnotnull(product_id#4))
            :     :  +- *(3) ColumnarToRow
            :     :     +- BatchScan parquet /tables/orders[customer_id#6, product_id#4, amount#3] ParquetScan DataFilters: [isnotnull(customer_id#6), isnotnull(product_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id), IsNotNull(product_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,product_id:int,amount:int> RuntimeFilters: []
            :     +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=N]
            :        +- *(1) Filter isnotnull(customer_id#7)
            :           +- *(1) ColumnarToRow
            :              +- BatchScan parquet /tables/customers[customer_id#7, tier#1] ParquetScan DataFilters: [isnotnull(customer_id#7)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []
            +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=N]
               +- *(2) Filter isnotnull(product_id#5)
                  +- *(2) ColumnarToRow
                     +- BatchScan parquet /tables/products[product_id#5, category#2] ParquetScan DataFilters: [isnotnull(product_id#5)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(product_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<product_id:int,category:string> RuntimeFilters: []

It returns

tier	category	total
free	cat_0	84774
free	cat_2	84200
free	cat_4	83626
free	cat_6	84410
pro	cat_1	84487
pro	cat_3	83913
pro	cat_5	84018
pro	cat_7	84802

The answer most people give

"Three shuffles — two joins and a group by." Only sort-merge joins shuffle. Counting joins rather than reading strategies is the habit that produces this answer, and the plan disproves it in one line.

They’ll ask next

One dimension grows to 2 GB. What does the plan look like, and what would you change in the model rather than the config?

Joins in Spark Broadcast Joins, Shared Variables & Bucketing

10 / 21

Predicate & projection pushdownJoin strategies (broadcast / SMJ / skew join / salting)

The filter on region is written after the join, but the plan shows it in the scan. What did Catalyst do, and when will it refuse to do it?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

The code — what does Spark do with it?

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .filter(F.col("region") == "EU")
        .select("order_id", "tier", "amount"))

The code — predict the output before reading on

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .filter(F.col("region") == "EU")
        .select("order_id", "tier", "amount"))

Why they ask this

Predicate pushdown is the optimisation everyone names and few can bound. Knowing when it does *not* happen is what separates the two.

Say this

It pushed the predicate through the join and into the Parquet scan, so region never has to reach the join at all. It refuses when the predicate is not deterministic or the optimizer cannot prove the reorder is safe — a UDF, or the null-producing side of an outer join.

The reasoning

Catalyst rewrites the logical plan before choosing physical operators, and pushing filters toward the source is one of the first rules it applies. The predicate ends up in the scan's PushedFilters, which means Parquet itself skips row groups whose statistics say they cannot contain a match — the rows are never decoded, let alone joined.

So filter placement in your code is largely a readability choice for simple predicates, and that is worth saying because a lot of hand-optimisation is spent on it. Where it stops being free is the boundary: Catalyst will not push a predicate it cannot reason about, and a Python UDF is the clearest case — it is an opaque function, so the plan shows BatchEvalPython and the scan reads everything.

The other case worth naming is outer joins. A filter on the null-producing side cannot be pushed below the join, because after an outer join those columns may be null for rows that had no match, and filtering earlier would remove rows the join was supposed to preserve. That is the standard trap in 'why is my left join behaving like an inner join' — the filter in the WHERE clause turned it into one.

What it actually returns 0 shuffles, run on Spark 4.2

Written after the join, executed in the scan.

The physical plan0 shuffles

== Physical Plan ==
*(2) Project [order_id#1, tier#2, amount#3]
+- *(2) BroadcastHashJoin [customer_id#4], [customer_id#5], Inner, BuildRight, false, false
   :- *(2) Project [order_id#1, customer_id#4, amount#3]
   :  +- *(2) Filter ((isnotnull(region#6) AND (region#6 = EU)) AND isnotnull(customer_id#4))
   :     +- *(2) ColumnarToRow
   :        +- BatchScan parquet /tables/orders[order_id#1, customer_id#4, region#6, amount#3] ParquetScan DataFilters: [isnotnull(region#6), (region#6 = EU), isnotnull(customer_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(region), EqualTo(region,EU), IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<order_id:int,customer_id:int,region:string,amount:int> RuntimeFilters: []
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=N]
      +- *(1) Filter isnotnull(customer_id#5)
         +- *(1) ColumnarToRow
            +- BatchScan parquet /tables/customers[customer_id#5, tier#2] ParquetScan DataFilters: [isnotnull(customer_id#5)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []

It returns

order_id	tier	amount
0	free	5
3	pro	26
6	free	47
9	pro	68
12	free	89
15	pro	110
18	free	131
21	pro	152
24	free	173
27	pro	194
30	free	215
33	pro	236

The answer most people give

"You should always filter before joining for performance." For a simple predicate the optimizer already did it, and the plan proves it. The advice is not wrong, it is just not where the wins are — the wins are in what the optimizer cannot move.

They’ll ask next

Rewrite this as a left join with the same WHERE clause. What happens to the row count, and why?

Spark SQL & the Catalyst Optimizer Joins in Spark

11 / 21

Predicate & projection pushdownLazy evaluation & Catalyst

wide_events has 22 columns and the job selects three. Where in the plan can you see that Spark will not read the other nineteen?

The setup — predict the table before reading on

The tables

wide_events(event_id int, region string, c0 int, c1 int, c2 int, c3 int, c4 int, c5 int, c6 int, c7 int, c8 int, c9 int, c10 int, c11 int, c12 int, c13 int, c14 int, c15 int, c16 int, c17 int, c18 int, c19 int)

22 columns. A report needs three of them.

wide_events

event_id	region	c0	c1	c2	c3	c4	c5	c6	c7	c8	c9	c10	c11	c12	c13	c14	c15	c16	c17	c18	c19
0	EU	0	1	2	3	4	5	6	7	8	9	10	11	12	13	14	15	16	17	18	19
1	US	1	2	3	4	5	6	7	8	9	10	11	12	13	14	15	16	17	18	19	20
2	APAC	2	3	4	5	6	7	8	9	10	11	12	13	14	15	16	17	18	19	20	21
3	EU	3	4	5	6	7	8	9	10	11	12	13	14	15	16	17	18	19	20	21	22

The code — what does Spark do with it?

out = (spark.table("wide_events")
        .select("event_id", "region", "c0")
        .filter(F.col("region") == "EU"))

The code — predict the output before reading on

out = (spark.table("wide_events")
        .select("event_id", "region", "c0")
        .filter(F.col("region") == "EU"))

Why they ask this

Column pruning is the cheapest large win in a columnar format, and the plan states it explicitly. Candidates who have never read a ReadSchema line miss it entirely.

Say this

In the scan's ReadSchema, which lists only the columns the scan will decode. With Parquet those columns are stored separately on disk, so the other nineteen are never touched.

The reasoning

ReadSchema: struct<event_id:int,region:string,c0:int> is the line. It is not a projection applied after reading — it is what the scan is configured to read, and in a columnar format the remaining columns are physically elsewhere in the file and simply not fetched. On a wide table this is often a larger saving than any predicate pushdown.

This is also the concrete reason SELECT * is expensive in a way it is not in a row-oriented database. Every column you name widens ReadSchema; every one you do not is free. On a table like this, selecting three of twenty-two columns reads roughly an eighth of the bytes, and that ratio lands directly on a per-byte-scanned bill.

It composes with the predicate in the same scan: PushedFilters lets Parquet skip whole row groups using column statistics, and ReadSchema limits which columns are decoded from the groups that survive. Reading both lines off a scan tells you most of what a job will cost before you run it, which is the practical skill this question is testing.

What it actually returns 0 shuffles, run on Spark 4.2

Twenty-two columns in the table, three in ReadSchema.

The physical plan0 shuffles

== Physical Plan ==
*(1) Filter (isnotnull(region#1) AND (region#1 = EU))
+- *(1) ColumnarToRow
   +- BatchScan parquet /tables/wide_events[event_id#2, region#1, c0#3] ParquetScan DataFilters: [isnotnull(region#1), (region#1 = EU)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(region), EqualTo(region,EU)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<event_id:int,region:string,c0:int> RuntimeFilters: []

It returns

event_id	region	c0
0	EU	0
3	EU	3
6	EU	6
9	EU	9
12	EU	12
15	EU	15
18	EU	18
21	EU	21
24	EU	24
27	EU	27
30	EU	30
33	EU	33

The answer most people give

"Spark reads the file and then projects." That is true of a row format like CSV and false of Parquet, where columns are stored contiguously and the reader fetches only the ones requested.

They’ll ask next

The same job on gzipped CSV instead of Parquet. What changes in the plan and in the bytes read?

Reading & Writing — Sources & File Formats Spark SQL & the Catalyst Optimizer

12 / 21

Predicate & projection pushdownPartitioning, repartition vs coalesce

daily_orders is written partitioned by order_date and the job filters to one day. How is that different from the filter pushdown in the previous question?

The setup — predict the table before reading on

The tables

daily_orders(order_id int, order_date string, region string, amount int)

3,000 orders written partitioned by order_date, 30 day-partitions.

daily_orders

order_id	region	amount	order_date
0	EU	5	2026-03-01
1	US	12	2026-03-02
2	APAC	19	2026-03-03
3	EU	26	2026-03-04

The code — what does Spark do with it?

out = (spark.table("daily_orders")
        .filter(F.col("order_date") == "2026-03-05")
        .groupBy("region").agg(F.sum("amount").alias("total")))

The code — predict the output before reading on

out = (spark.table("daily_orders")
        .filter(F.col("order_date") == "2026-03-05")
        .groupBy("region").agg(F.sum("amount").alias("total")))

Why they ask this

The two get conflated constantly. One skips files without opening them; the other opens files and skips row groups inside them.

Say this

Partition pruning happens in the file listing — Spark never opens the other 29 days' directories. Filter pushdown happens inside a file that has been opened, using column statistics to skip row groups.

The reasoning

Partition pruning is a directory-level decision. The table is laid out as order_date=2026-03-01/, order_date=2026-03-02/ and so on, so a predicate on that column is answered by listing directories — 29 of the 30 are never opened, never read, and never counted in the bytes scanned. It is the single most effective layout decision you can make.

Filter pushdown is a within-file decision. The file is opened, and Parquet's per-row-group min/max statistics let the reader skip groups that cannot contain a match. It saves decode time and it still costs the file open and the footer read, which is why it is a smaller win than pruning and why the small-files problem hurts even with good filters.

The practical consequence is that you partition by what queries filter on *ranges* of — usually a date — and rely on pushdown plus clustering for everything else. And the classic way to lose pruning is to apply a function to the partition column in the predicate: a cast or a date format around it can leave Spark unable to match the filter to the partition values, and the plan quietly shows an empty PartitionFilters with every directory listed.

What it actually returns 1 shuffle, run on Spark 4.2

One day's directory, out of thirty on disk.

The physical plan1 shuffle

== Physical Plan ==
*(2) HashAggregate(keys=[region#1], functions=[sum(amount#2)])
+- Exchange hashpartitioning(region#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(1) HashAggregate(keys=[region#1], functions=[partial_sum(amount#2)])
      +- *(1) ColumnarToRow
         +- BatchScan parquet /tables/daily_orders[region#1, amount#2] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [isnotnull(order_date#3), (order_date#3 = 2026-03-05)], PushedAggregation: [], PushedFilters: [], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<region:string,amount:int> RuntimeFilters: []

It returns

region	total
US	33806

The answer most people give

"They are the same thing, one is just at the file level." They are different mechanisms with different costs, and only one of them avoids opening the file at all — which is what matters when the alternative is opening tens of thousands.

They’ll ask next

Someone writes filter(to_date(col('order_date')) == lit('2026-03-05')). What happens to the pruning?

Reading & Writing — Sources & File Formats Partitioning & Parallelism

13 / 21

UDFs vs native vs pandas UDFsPredicate & projection pushdown

The filter is a Python UDF returning amount > 300. What does the plan show, and what has it cost beyond the obvious serialisation?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

The code — what does Spark do with it?

big = F.udf(lambda a: a > 300, "boolean")
out = spark.table("orders").filter(big(F.col("amount"))).select("order_id", "amount")

The code — predict the output before reading on

big = F.udf(lambda a: a > 300, "boolean")
out = spark.table("orders").filter(big(F.col("amount"))).select("order_id", "amount")

Why they ask this

Everyone knows UDFs are slow. The interviewer wants the plan-level reason, which is that the optimizer stops being able to see what you are doing.

Say this

BatchEvalPython appears in the tree, and the filter is gone from the scan. The cost is not just serialising rows to Python — it is that every row must now be read and decoded, because nothing can be pushed down.

The reasoning

A Python UDF is opaque to Catalyst. It cannot know the function is equivalent to amount > 300, so it cannot push it into the scan, cannot use it for partition pruning, and cannot reorder it with anything. The plan shows BatchEvalPython as its own node, and the scan above it reads everything.

So there are two costs and the second is usually larger. The visible one is leaving the JVM: rows are serialised to Python, evaluated a batch at a time, and the results serialised back. The invisible one is the optimisation that no longer happens — the row-group skipping that a native predicate would have got for free, on a table where the filter might have eliminated most of the data before it was ever decoded.

The fix, in order of preference: express it natively, because F.col('amount') > 300 is pushed down and runs in the JVM; if it genuinely needs Python, use a pandas UDF, which is vectorised over Arrow batches and removes most of the serialisation cost though not the opacity; and if you must have a scalar Python UDF, apply it as late as possible so it sees the fewest rows.

What it actually returns 0 shuffles, run on Spark 4.2

BatchEvalPython in the tree, and nothing in PushedFilters.

The physical plan0 shuffles

== Physical Plan ==
*(2) Project [order_id#1, amount#2]
+- *(2) Filter pythonUDF0#3: boolean
   +- BatchEvalPython [<lambda>(amount#2)#4], [pythonUDF0#3]
      +- *(1) ColumnarToRow
         +- BatchScan parquet /tables/orders[order_id#1, amount#2] ParquetScan DataFilters: [<lambda>(amount#2)#4], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<order_id:int,amount:int> RuntimeFilters: []

It returns

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

The answer most people give

"UDFs are slow because of Python." Partly — and the bigger cost on a large table is that the scan can no longer skip anything, so you read and decode every row before Python even sees it.

They’ll ask next

Rewrite it as a pandas UDF. Which of the two costs does that remove?

UDFs & the pandas API on Spark Spark SQL & the Catalyst Optimizer

14 / 21

Predicate & projection pushdownPartitioning, repartition vs coalesce

The same predicate as a native column expression, with and without a repartition in front of it. Compare the two plans.

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

The code — what does Spark do with it?

out = spark.table("orders").filter(F.col("amount") > 300).select("order_id", "amount")

The job as written — it works, and it is slow

out = (spark.table("orders")
        .repartition(20)
        .filter(F.col("amount") > 300)
        .select("order_id", "amount"))

The physical plan1 shuffle

== Physical Plan ==
Exchange RoundRobinPartitioning(20), REPARTITION_BY_NUM, [plan_id=N]
+- *(1) Filter (isnotnull(amount#1) AND (amount#1 > 300))
   +- *(1) ColumnarToRow
      +- BatchScan parquet /tables/orders[order_id#2, amount#1] ParquetScan DataFilters: [isnotnull(amount#1), (amount#1 > 300)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(amount), GreaterThan(amount,300)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<order_id:int,amount:int> RuntimeFilters: []

It returns

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

The code — predict the output before reading on

out = spark.table("orders").filter(F.col("amount") > 300).select("order_id", "amount")

Why they ask this

It pairs the UDF question and adds the repartition trap, so the candidate has to read two differences rather than recognise one.

Say this

Natively the filter reaches PushedFilters and there is no Python in the tree at all. The repartition adds an Exchange that buys nothing — same rows, one extra shuffle, which the harness asserts.

The reasoning

The native version is what you want a scan to look like: the predicate appears in both DataFilters and PushedFilters, so Parquet skips row groups, and there is no separate Filter node doing work the reader already did. No BatchEvalPython, no serialisation, nothing between the file and the result.

The repartition in the as-written version is the second lesson. It produces an Exchange, and nothing downstream needs a particular distribution — there is no join and no aggregation — so the shuffle is pure cost. The two versions return identical rows and differ by one full network shuffle, which is why the pair is published together.

The instinct behind it is usually 'the job only used two cores, so let me increase parallelism'. That is a real problem with a different fix: parallelism on a scan is set by the number of input splits, so the answer is file layout — more, appropriately-sized files, or a smaller maxPartitionBytes — not a shuffle after the fact.

What it actually returns 0 shuffles, run on Spark 4.2

Same rows either way. One version pays for a shuffle.

The physical plan0 shuffles

== Physical Plan ==
*(1) Filter (isnotnull(amount#1) AND (amount#1 > 300))
+- *(1) ColumnarToRow
   +- BatchScan parquet /tables/orders[order_id#2, amount#1] ParquetScan DataFilters: [isnotnull(amount#1), (amount#1 > 300)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(amount), GreaterThan(amount,300)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<order_id:int,amount:int> RuntimeFilters: []

It returns

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

The answer most people give

"repartition after reading increases parallelism, so it is worth the shuffle." Scan parallelism comes from input splits, and a shuffle to raise it costs a full pass over the network to fix something file layout should have fixed.

They’ll ask next

The scan produces two partitions and you want twenty. What do you change?

Reading & Writing — Sources & File Formats Partitioning & Parallelism

15 / 21

Shuffle mechanicsLazy evaluation & Catalyst

.select("region", "customer_id").distinct() — what does the physical plan look like, and is it different from a groupBy on those two columns?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

The code — what does Spark do with it?

out = spark.table("orders").select("region", "customer_id").distinct()

The code — predict the output before reading on

out = spark.table("orders").select("region", "customer_id").distinct()

Why they ask this

It checks whether the candidate reads plans rather than APIs. Two different-looking pieces of code compiling to the same plan is a useful thing to have noticed.

Say this

It compiles to exactly the aggregation shape: partial HashAggregate, Exchange, final HashAggregate — the same plan a groupBy on both columns with no aggregate function produces.

The reasoning

distinct() is grouping by every selected column with no aggregate expression, and the physical plan says so: HashAggregate with keys and empty functions, an Exchange between the partial and final passes, and nothing else. Writing groupBy('region','customer_id').count().drop('count') would produce the same tree with one more expression.

That matters because it tells you what distinct costs. It is a full shuffle on the distinct columns, and its cost scales with the *cardinality* of the combination — the number of distinct values is what has to be held in memory per partition and shipped. Two low-cardinality columns are cheap; adding a high-cardinality id makes it as expensive as any wide aggregation.

The version to be careful with is countDistinct inside a wider aggregation, because it cannot use the partial-aggregate trick in the same way — Spark has to get all values for a key onto one partition to count them uniquely. That is why approx_count_distinct exists, and why swapping it in is a legitimate tuning move when an exact answer is not required.

What it actually returns 1 shuffle, run on Spark 4.2

HashAggregate with keys and no functions. That is what distinct is.

The physical plan1 shuffle

== Physical Plan ==
*(2) HashAggregate(keys=[region#1, customer_id#2], functions=[])
+- Exchange hashpartitioning(region#1, customer_id#2, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(1) HashAggregate(keys=[region#1, customer_id#2], functions=[])
      +- *(1) ColumnarToRow
         +- BatchScan parquet /tables/orders[customer_id#2, region#1] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,region:string> RuntimeFilters: []

It returns

region	customer_id
APAC	0
APAC	1
APAC	2
APAC	3
APAC	4
APAC	5
APAC	6
APAC	7
APAC	8
APAC	9
APAC	10
APAC	11

The answer most people give

"distinct is cheaper than a group by because it does not aggregate anything." It is the same plan. The shuffle is the cost, and it does not care whether there is a sum on top.

They’ll ask next

Where does countDistinct differ, and what would you use instead at scale?

Aggregations & Grouping The Shuffle — Spark's Performance Killer

16 / 21

AQELazy evaluation & Catalyst

With spark.sql.adaptive.enabled=false, the plan starts straight at HashAggregate. What does that tell you about when the plan was decided?

The setup — predict the table before reading on

spark.sql.adaptive.enabledfalse

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

The code — what does Spark do with it?

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .groupBy("tier").agg(F.sum("amount").alias("total")))

The code — predict the output before reading on

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .groupBy("tier").agg(F.sum("amount").alias("total")))

Why they ask this

It sets up the contrast with AQE, and it makes the candidate say when planning happens relative to execution — which is the whole basis of lazy evaluation.

Say this

It was decided entirely before any data moved, from statistics estimated off file sizes. What you see is what will run, and it cannot change however wrong the estimates turn out to be.

The reasoning

Without AQE, Catalyst plans once. It estimates sizes from whatever statistics exist — here, Parquet file sizes, since nothing has been ANALYZEd — picks join strategies and partition counts from those estimates, and commits. The plan you print is the plan that executes.

That is fine when the estimates are good and poor when they are not, and after a few filters and joins they are frequently not. A filter Catalyst thinks removes 10% might remove 99%, and it will still shuffle into 200 partitions and still choose a sort-merge join for a side that turned out to be tiny. Nothing observes the actual data and nothing corrects course.

It is also why spark.sql.shuffle.partitions was such a blunt instrument for so long. It is one number applied to every shuffle in the application, tuned for the largest stage and therefore wasteful for every other one — 200 near-empty tasks after a selective filter, each with real scheduling overhead.

What it actually returns 1 shuffle, run on Spark 4.2

No AdaptiveSparkPlan wrapper. This tree is final.

The physical plan1 shuffle

== Physical Plan ==
*(3) HashAggregate(keys=[tier#1], functions=[sum(amount#2)])
+- Exchange hashpartitioning(tier#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(2) HashAggregate(keys=[tier#1], functions=[partial_sum(amount#2)])
      +- *(2) Project [amount#2, tier#1]
         +- *(2) BroadcastHashJoin [customer_id#3], [customer_id#4], Inner, BuildRight, false, false
            :- *(2) Filter isnotnull(customer_id#3)
            :  +- *(2) ColumnarToRow
            :     +- BatchScan parquet /tables/orders[customer_id#3, amount#2] ParquetScan DataFilters: [isnotnull(customer_id#3)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,amount:int> RuntimeFilters: []
            +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=N]
               +- *(1) Filter isnotnull(customer_id#4)
                  +- *(1) ColumnarToRow
                     +- BatchScan parquet /tables/customers[customer_id#4, tier#1] ParquetScan DataFilters: [isnotnull(customer_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []

It returns

tier	total
free	337010
pro	337220

The answer most people give

"Spark always optimises at runtime." Only with AQE enabled. With it off, every decision in that tree was made from estimates before a single byte was read.

They’ll ask next

Which of those decisions would you most want revisited at runtime, and why?

Adaptive Query Execution & Cost-Based Optimization Spark SQL & the Catalyst Optimizer

17 / 21

AQELazy evaluation & Catalyst

With AQE enabled the same query prints AdaptiveSparkPlan isFinalPlan=false at the root. What does that flag mean, and what will Spark be allowed to change?

The setup — predict the table before reading on

spark.sql.adaptive.enabledtrue

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

customers(customer_id int, tier string, country string)

400 customers. Small enough to broadcast.

customers

customer_id	tier	country
0	free	EU
1	pro	US
2	free	APAC
3	pro	EU

The code — what does Spark do with it?

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .groupBy("tier").agg(F.sum("amount").alias("total")))

The code — predict the output before reading on

out = (spark.table("orders")
        .join(spark.table("customers"), "customer_id")
        .groupBy("tier").agg(F.sum("amount").alias("total")))

Why they ask this

Reading isFinalPlan=false correctly is a small thing that tells an interviewer you have looked at modern plans rather than learned Spark 2 output.

Say this

It means what you are looking at is provisional — the query has not run, so Spark has no runtime statistics yet. Once each stage completes it can coalesce partitions, switch a join to a broadcast, and split skewed partitions.

The reasoning

AQE re-plans at stage boundaries. A shuffle produces exact statistics about what was actually written, and AQE uses them to make decisions the pre-run estimates could not. Until the query runs, the printed tree is a starting point, which is precisely what the flag is telling you.

Three things it can change, and they are worth naming individually because they are separately configurable. It coalesces post-shuffle partitions, so a selective filter no longer leaves you with 200 tiny tasks. It converts a sort-merge join to a broadcast when the shuffled side turns out to fit — the win the manual hint gives you, without hard-coding a size. And it splits skewed partitions, which is the automatic half of the salting answer.

To see the plan that actually ran, print after an action, or read the SQL tab in the Spark UI — both show isFinalPlan=true with the rewritten tree. That is a genuinely useful debugging habit: comparing the provisional plan against the final one shows you exactly which of your estimates were wrong.

What it actually returns 1 shuffle, run on Spark 4.2

Same query as the previous question, AQE enabled.

The physical plan1 shuffle

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[tier#1], functions=[sum(amount#2)])
   +- Exchange hashpartitioning(tier#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
      +- HashAggregate(keys=[tier#1], functions=[partial_sum(amount#2)])
         +- Project [amount#2, tier#1]
            +- BroadcastHashJoin [customer_id#3], [customer_id#4], Inner, BuildRight, false, false
               :- Filter isnotnull(customer_id#3)
               :  +- BatchScan parquet /tables/orders[customer_id#3, amount#2] ParquetScan DataFilters: [isnotnull(customer_id#3)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,amount:int> RuntimeFilters: []
               +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=N]
                  +- Filter isnotnull(customer_id#4)
                     +- BatchScan parquet /tables/customers[customer_id#4, tier#1] ParquetScan DataFilters: [isnotnull(customer_id#4)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(customer_id)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<customer_id:int,tier:string> RuntimeFilters: []

It returns

tier	total
free	337010
pro	337220

The answer most people give

"isFinalPlan=false means the plan failed or is incomplete." It means the query has not executed yet, so no runtime statistics exist. It is the normal state of an adaptive plan printed before an action.

They’ll ask next

How do you see the final plan, and what would you compare it against?

Adaptive Query Execution & Cost-Based Optimization Performance Tuning & Debugging with the Spark UI

18 / 21

Shuffle mechanicsWide vs narrow transformations

A row_number over a window partitioned by region and ordered by amount, filtered to the top 2. What does the plan add, and why can the filter not be pushed below it?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

The code — what does Spark do with it?

from pyspark.sql.window import Window
w = Window.partitionBy("region").orderBy(F.col("amount").desc())
out = (spark.table("orders")
        .withColumn("rn", F.row_number().over(w))
        .filter(F.col("rn") <= 2)
        .select("region", "order_id", "amount", "rn"))

The code — predict the output before reading on

from pyspark.sql.window import Window
w = Window.partitionBy("region").orderBy(F.col("amount").desc())
out = (spark.table("orders")
        .withColumn("rn", F.row_number().over(w))
        .filter(F.col("rn") <= 2)
        .select("region", "order_id", "amount", "rn"))

Why they ask this

Top-N-per-group is one of the most common real jobs, and the plan shows both costs people forget: the shuffle and the sort.

Say this

An Exchange to bring each region together and a Sort within each partition — windows require both. The rank filter cannot be pushed below the window because the rank does not exist until the window has been computed.

The reasoning

A window is a wide operation. partitionBy('region') requires every row for a region on one partition, which is an Exchange; orderBy requires those rows sorted within the partition, which is a Sort. Then the window operator walks each partition assigning row numbers. Both costs are visible in the tree and neither is avoidable for this shape.

The filter on rn <= 2 sits above the window and cannot move below it, because rn is produced by the window — there is nothing to filter on until it has run. That is the structural difference from a plain predicate, and it is why top-N per group reads all the rows even though you only want two per group.

The cost concentrates on whichever partition key is skewed, which makes this pattern a common source of one-slow-task incidents: if a region holds most of the rows, one task sorts most of the data. AQE's skew handling helps with joins and not with window partitions, so the fixes there are structural — pre-aggregate if the window is over an aggregate, or narrow the window key.

What it actually returns 1 shuffle, run on Spark 4.2

Exchange plus Sort. Both are required, neither is optional.

The physical plan1 shuffle

== Physical Plan ==
*(3) Project [region#1, order_id#2, amount#3, rn#4]
+- *(3) Filter (rn#4 <= 2)
   +- Window [row_number() windowspecdefinition(region#1, amount#3 DESC NULLS LAST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS rn#4], [region#1], [amount#3 DESC NULLS LAST]
      +- WindowGroupLimit [region#1], [amount#3 DESC NULLS LAST], row_number(), 2, Final
         +- *(2) Sort [region#1 ASC NULLS FIRST, amount#3 DESC NULLS LAST], false, 0
            +- Exchange hashpartitioning(region#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
               +- WindowGroupLimit [region#1], [amount#3 DESC NULLS LAST], row_number(), 2, Partial
                  +- *(1) Sort [region#1 ASC NULLS FIRST, amount#3 DESC NULLS LAST], false, 0
                     +- *(1) ColumnarToRow
                        +- BatchScan parquet /tables/orders[order_id#2, region#1, amount#3] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<order_id:int,region:string,amount:int> RuntimeFilters: []

It returns

region	order_id	amount	rn
APAC	290	677	1
APAC	581	677	2
EU	96	677	1
EU	387	677	2
US	193	677	1
US	484	677	2

The answer most people give

"Filtering rn <= 2 means Spark only reads two rows per region." The rank has to be computed before it can be filtered, so every row is shuffled and sorted first. The filter reduces the output, not the work.

They’ll ask next

One region is 80% of the rows. What happens, and what would you do about it?

Window Functions The Shuffle — Spark's Performance Killer

19 / 21

Lazy evaluation & CatalystShuffle mechanics

Two filters on the same table, unioned, then aggregated. How many times does Spark scan the table, and what does that tell you about reuse?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

The code — what does Spark do with it?

o = spark.table("orders")
eu = o.filter(F.col("region") == "EU")
us = o.filter(F.col("region") == "US")
out = eu.union(us).groupBy("region").agg(F.count("*").alias("n"))

The code — predict the output before reading on

o = spark.table("orders")
eu = o.filter(F.col("region") == "EU")
us = o.filter(F.col("region") == "US")
out = eu.union(us).groupBy("region").agg(F.count("*").alias("n"))

Why they ask this

It probes whether the candidate understands that a DataFrame is a plan, not a result — and that referring to it twice can mean reading twice.

Say this

Twice. The union has two scan branches in the plan, because a DataFrame is a recipe and using it in two places re-executes it — there is no automatic caching of intermediate results.

The reasoning

The plan has two BatchScan nodes over the same table, one per branch of the union. Nothing is reused, because a DataFrame is a description of how to compute something rather than a materialised result — asking for it twice asks for the computation twice.

That is the practical face of lazy evaluation and the most common surprise it produces. A branch used in five places is read five times; a chain with an expensive shuffle feeding three outputs does the shuffle three times. The Spark UI shows it as more jobs and more input bytes than the code seems to justify.

The remedies are cache()/persist() when the branch is genuinely reused and expensive, and rewriting to avoid the fan-out when it is not. Here, the whole union collapses to a single filter — region IN ('EU','US') — which reads the table once and is what you would actually ship. Reaching for cache before checking whether the query can be rewritten is a common and expensive habit.

What it actually returns 1 shuffle, run on Spark 4.2

Two scans of one table, from two references to one DataFrame.

The physical plan1 shuffle

== Physical Plan ==
*(4) HashAggregate(keys=[region#1], functions=[count(1)])
+- Exchange hashpartitioning(region#1, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(3) HashAggregate(keys=[region#1], functions=[partial_count(1)])
      +- Union
         :- *(1) Filter (isnotnull(region#1) AND (region#1 = EU))
         :  +- *(1) ColumnarToRow
         :     +- BatchScan parquet /tables/orders[region#1] ParquetScan DataFilters: [isnotnull(region#1), (region#1 = EU)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(region), EqualTo(region,EU)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<region:string> RuntimeFilters: []
         +- *(2) Filter (isnotnull(region#2) AND (region#2 = US))
            +- *(2) ColumnarToRow
               +- BatchScan parquet /tables/orders[region#2] ParquetScan DataFilters: [isnotnull(region#2), (region#2 = US)], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(region), EqualTo(region,US)], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<region:string> RuntimeFilters: []

It returns

region	n
EU	667
US	667

The answer most people give

"Spark caches the DataFrame automatically after the first use." It does not. Nothing is retained between actions unless you ask for it, which is why re-reading is the default behaviour rather than the exception.

They’ll ask next

Rewrite this so the table is scanned once, without using cache.

Transformations, Actions & Lazy Evaluation Spark SQL & the Catalyst Optimizer

20 / 21

Shuffle mechanicsWide vs narrow transformations

An aggregation followed by orderBy on the total. The plan has two Exchanges — what is the second one doing, and how is it different from the first?

The setup — predict the table before reading on

The tables

orders(order_id int, customer_id int, product_id int, region string, order_date string, amount int)

2,000 orders across 3 regions and 30 days.

orders

order_id	customer_id	product_id	region	order_date	amount
0	0	0	EU	2026-03-01	5
1	1	1	US	2026-03-02	12
2	2	2	APAC	2026-03-03	19
3	3	3	EU	2026-03-04	26

The code — what does Spark do with it?

out = (spark.table("orders")
        .groupBy("region").agg(F.sum("amount").alias("total"))
        .orderBy(F.col("total").desc()))

The code — predict the output before reading on

out = (spark.table("orders")
        .groupBy("region").agg(F.sum("amount").alias("total"))
        .orderBy(F.col("total").desc()))

Why they ask this

Global ordering is requested casually and costs a full shuffle with a special partitioner. Most candidates cannot say what rangepartitioning is.

Say this

The second Exchange is a rangepartitioning, not a hashpartitioning: a global sort needs partitions that are ordered relative to each other, so Spark samples the data to pick range boundaries first.

The reasoning

A global orderBy cannot be satisfied by sorting each partition independently — partition 3 must contain values that all come after partition 2's. Spark achieves that with range partitioning: it samples the data to estimate the distribution, picks boundaries, shuffles rows into the right range, and sorts within each partition. The plan shows Exchange rangepartitioning(...), which is the tell.

That sampling pass is real work and it is the reason a global sort costs more than its shuffle alone suggests. It is also why skew hurts here — if the sort key is concentrated, the sampled boundaries produce unbalanced ranges and one task does most of the sorting.

So the useful question is always whether you need a *global* order. If the result is going to a file, sortWithinPartitions gives you ordering inside each output file at a fraction of the cost. If you only want the top few rows, orderBy(...).limit(n) lets Spark do a partial top-N instead of a full sort. Sorting an entire result set that a BI tool will re-sort anyway is pure waste, and it is extremely common.

What it actually returns 2 shuffles, run on Spark 4.2

hashpartitioning for the aggregate, rangepartitioning for the sort.

The physical plan2 shuffles

== Physical Plan ==
*(3) Sort [total#1L DESC NULLS LAST], true, 0
+- Exchange rangepartitioning(total#1L DESC NULLS LAST, 200), ENSURE_REQUIREMENTS, [plan_id=N]
   +- *(2) HashAggregate(keys=[region#2], functions=[sum(amount#3)])
      +- Exchange hashpartitioning(region#2, 200), ENSURE_REQUIREMENTS, [plan_id=N]
         +- *(1) HashAggregate(keys=[region#2], functions=[partial_sum(amount#3)])
            +- *(1) ColumnarToRow
               +- BatchScan parquet /tables/orders[region#2, amount#3] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[...], PartitionFilters: [], PushedAggregation: [], PushedFilters: [], PushedGroupBy: [], PushedVariantExtractions: [], ReadSchema: struct<region:string,amount:int> RuntimeFilters: []

It returns

region	total
APAC	224376
EU	224969
US	224885

The answer most people give

"orderBy is cheap because the aggregated result is small." The result is small and the mechanism is still a sampling pass plus a full shuffle with a range partitioner. Cheap here, and the same code on a large result is not.

They’ll ask next

You are writing this to Parquet. What do you use instead of orderBy, and what do you give up?

The Shuffle — Spark's Performance Killer Partitioning & Parallelism

21 / 21

EvergreenLazy evaluation & CatalystJoin strategies (broadcast / SMJ / skew join / salting)

What is the difference between a logical plan and a physical plan, and what does explain(True) show you that explain() does not?

Why they ask this

It is the vocabulary check before any real plan-reading question, and the extended output is where a candidate proves they have actually looked rather than read about it.

Say this

The logical plan is what the query means; the physical plan is how Spark will run it. explain(True) prints all four stages — parsed, analyzed, optimized and physical — so you can see which rewrite happened where.

The reasoning

The **logical plan** describes the operations and their order without committing to how any of them run. A join is just a join. This is the level at which Catalyst applies its safe rewrites: pushing a filter under a join, pruning unread columns, folding constants. Comparing the parsed plan to the optimized one tells you exactly which rewrites fired.

The **physical plan** is the executable one. Here a join has become `BroadcastHashJoin` or `SortMergeJoin`, an aggregation has become `HashAggregate`, and shuffles appear as explicit `Exchange` nodes. This is where the strategy decisions live, and it is the level almost every plan question is actually about — count the Exchanges, name the join, check `PushedFilters` and `ReadSchema` on the scan.

`explain()` prints only the physical plan. `explain(True)` prints all four sections: parsed logical, analyzed logical, optimized logical, and physical. That is what you want when the question is *why* Spark chose something, because the answer is usually visible as a difference between two of those sections.

The distinction that matters in practice: the logical rewrites are always safe, so they do not depend on statistics. The physical choice does — it is cost-based, and a stale or missing row count is what makes Spark broadcast a table that turns out to be far larger than it estimated. So "the plan changed and my code did not" is nearly always a physical-plan story, not a logical one.

The formulations

explain(True) — all fourship

df.explain(True)
# == Parsed Logical Plan ==  / == Analyzed ==
# == Optimized Logical Plan == / == Physical Plan ==

The only way to see which rewrite happened at which stage.

explain() — physical onlyworks

df.explain()
# == Physical Plan ==  Exchange / SortMergeJoin / FileScan

Enough for "how many shuffles" and "which join". Not enough for "why".

explain("cost")works

df.explain('cost')   # statistics attached to logical nodes

Shows the size estimates that drove the physical choice.

The answer most people give

"The logical plan is what you wrote and the physical plan is optimized." Both are optimized, at different levels — rule-based rewrites on the logical side, cost-based strategy selection on the physical. The parsed logical plan is the one that is closest to what you wrote.

They’ll ask next

Your code has not changed but the plan now shows SortMergeJoin instead of BroadcastHashJoin. Where do you look?

Spark SQL & the Catalyst Optimizer Adaptive Query Execution & Cost-Based Optimization

event_id	region	c0
0	EU	0
3	EU	3
6	EU	6
9	EU	9
12	EU	12
15	EU	15
18	EU	18
21	EU	21
24	EU	24
27	EU	27
30	EU	30
33	EU	33

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

event_id	region	c0
0	EU	0
3	EU	3
6	EU	6
9	EU	9
12	EU	12
15	EU	15
18	EU	18
21	EU	21
24	EU	24
27	EU	27
30	EU	30
33	EU	33

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

event_id	region	c0
0	EU	0
3	EU	3
6	EU	6
9	EU	9
12	EU	12
15	EU	15
18	EU	18
21	EU	21
24	EU	24
27	EU	27
30	EU	30
33	EU	33

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

event_id	region	c0
0	EU	0
3	EU	3
6	EU	6
9	EU	9
12	EU	12
15	EU	15
18	EU	18
21	EU	21
24	EU	24
27	EU	27
30	EU	30
33	EU	33

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

Counting shuffles & stages

Which join, and why

Where the filter ended up

AQE & the rest of the tree

Evergreen · asked verbatim

Filter orders to EU, join customers, group by tier. How many shuffles does Spark do, and which line of the code caused each one?

Same join and aggregation, but autoBroadcastJoinThreshold is -1 so nothing can be broadcast. How many shuffles now, and where did the Sort nodes come from?

This aggregates orders per customer and then joins the customer dimension. Is the shuffle count different from joining first and aggregating after, and why would you write it this way?

Someone added .repartition(50, "customer_id") before a groupBy on region, believing it would speed the aggregation up. What did it actually do to the plan?

Same code, but repartitioned on region — the column the groupBy uses. How many Exchanges are in the plan now?

This join produced a BroadcastHashJoin with BuildRight and zero shuffles. What did the optimizer know that made that decision, and what would take it away?

Identical code to the previous question, with autoBroadcastJoinThreshold set to 1 byte. What is the plan now, and what does that tell you about tuning by config?

Broadcast is disabled by the threshold, but F.broadcast() is applied to the small side. Which wins, and what have you taken responsibility for by writing that?

orders joined to customers joined to products, then grouped by tier and category. Two joins and an aggregation — why is there only one Exchange?

The filter on region is written after the join, but the plan shows it in the scan. What did Catalyst do, and when will it refuse to do it?

wide_events has 22 columns and the job selects three. Where in the plan can you see that Spark will not read the other nineteen?

daily_orders is written partitioned by order_date and the job filters to one day. How is that different from the filter pushdown in the previous question?

The filter is a Python UDF returning amount > 300. What does the plan show, and what has it cost beyond the obvious serialisation?

The same predicate as a native column expression, with and without a repartition in front of it. Compare the two plans.

.select("region", "customer_id").distinct() — what does the physical plan look like, and is it different from a groupBy on those two columns?

With spark.sql.adaptive.enabled=false, the plan starts straight at HashAggregate. What does that tell you about when the plan was decided?

With AQE enabled the same query prints AdaptiveSparkPlan isFinalPlan=false at the root. What does that flag mean, and what will Spark be allowed to change?

A row_number over a window partitioned by region and ordered by amount, filtered to the top 2. What does the plan add, and why can the filter not be pushed below it?

Two filters on the same table, unioned, then aggregated. How many times does Spark scan the table, and what does that tell you about reuse?

An aggregation followed by orderBy on the total. The plan has two Exchanges — what is the second one doing, and how is it different from the first?

What is the difference between a logical plan and a physical plan, and what does explain(True) show you that explain() does not?

Counting shuffles & stages

Which join, and why

Where the filter ended up

AQE & the rest of the tree

Evergreen · asked verbatim

Filter orders to EU, join customers, group by tier. How many shuffles does Spark do, and which line of the code caused each one?

Same join and aggregation, but autoBroadcastJoinThreshold is -1 so nothing can be broadcast. How many shuffles now, and where did the Sort nodes come from?

This aggregates orders per customer and then joins the customer dimension. Is the shuffle count different from joining first and aggregating after, and why would you write it this way?

Someone added .repartition(50, "customer_id") before a groupBy on region, believing it would speed the aggregation up. What did it actually do to the plan?

Same code, but repartitioned on region — the column the groupBy uses. How many Exchanges are in the plan now?

This join produced a BroadcastHashJoin with BuildRight and zero shuffles. What did the optimizer know that made that decision, and what would take it away?

Identical code to the previous question, with autoBroadcastJoinThreshold set to 1 byte. What is the plan now, and what does that tell you about tuning by config?

Broadcast is disabled by the threshold, but F.broadcast() is applied to the small side. Which wins, and what have you taken responsibility for by writing that?

orders joined to customers joined to products, then grouped by tier and category. Two joins and an aggregation — why is there only one Exchange?

The filter on region is written after the join, but the plan shows it in the scan. What did Catalyst do, and when will it refuse to do it?

wide_events has 22 columns and the job selects three. Where in the plan can you see that Spark will not read the other nineteen?

daily_orders is written partitioned by order_date and the job filters to one day. How is that different from the filter pushdown in the previous question?

The filter is a Python UDF returning amount > 300. What does the plan show, and what has it cost beyond the obvious serialisation?

The same predicate as a native column expression, with and without a repartition in front of it. Compare the two plans.

.select("region", "customer_id").distinct() — what does the physical plan look like, and is it different from a groupBy on those two columns?

With spark.sql.adaptive.enabled=false, the plan starts straight at HashAggregate. What does that tell you about when the plan was decided?

With AQE enabled the same query prints AdaptiveSparkPlan isFinalPlan=false at the root. What does that flag mean, and what will Spark be allowed to change?

A row_number over a window partitioned by region and ordered by amount, filtered to the top 2. What does the plan add, and why can the filter not be pushed below it?

Two filters on the same table, unioned, then aggregated. How many times does Spark scan the table, and what does that tell you about reuse?

An aggregation followed by orderBy on the total. The plan has two Exchanges — what is the second one doing, and how is it different from the first?

What is the difference between a logical plan and a physical plan, and what does explain(True) show you that explain() does not?

event_id	region	c0
0	EU	0
3	EU	3
6	EU	6
9	EU	9
12	EU	12
15	EU	15
18	EU	18
21	EU	21
24	EU	24
27	EU	27
30	EU	30
33	EU	33

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

event_id	region	c0
0	EU	0
3	EU	3
6	EU	6
9	EU	9
12	EU	12
15	EU	15
18	EU	18
21	EU	21
24	EU	24
27	EU	27
30	EU	30
33	EU	33

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383

order_id	amount
43	306
44	313
45	320
46	327
47	334
48	341
49	348
50	355
51	362
52	369
53	376
54	383