Read the Plan

An easy Spark interview practice problem on DataDriven. Write the Spark code and run it against the tables, with instant feedback.

Domain
Spark
Difficulty
easy
Seniority
L4
Asked in
Databricks

Problem

A 30 MB stores dimension should never cost an 80 GB shuffle, yet the platform team's order enrichment plan puts an Exchange on both sides of a SortMergeJoin and runs 12 minutes against a 4-minute SLA. Work out why the planner passed on the cheaper strategy, then rebuild the enrichment from retail_orders and stores so the orders side never moves, keeping each order's id and amount with its store's name and region.

30 MB table. 80 GB shuffle. Read the plan.

Expected output

First 5 of 200 rows
order_idamountstore_nameregion
6000326.29Store #002MIDWEST
6000647.59Store #003WEST
6000968.89Store #004SOUTHWEST
6001290.19Store #005NORTHEAST
60015111.49Store #006SOUTHEAST

Worked solution and explanation

This is an order-grain lookup that ended up as a cluster-wide reshuffle. The output never changes: one row per order, carrying its own store's name and region, with stores adding columns and never adding or dropping an order. What separates candidates is reading the WARN line: a 30 MB stores estimate lost to a 10 MB spark.sql.autoBroadcastJoinThreshold, so Catalyst fell back to SortMergeJoin and repartitioned all 80 GB of orders. Miss that and start tuning memory, and you keep the 80 GB shuffle and the missed SLA.

Read the plan before you touch the cluster

Step 1: Find the Exchange that moves the big side

Read the plan from the bottom up. Each Exchange hashpartitioning(store_id, 200) is a full network shuffle. One of them sits above a 500M-row scan, and that one node is the 12 minutes. The second Exchange, on stores, is 30 MB of wasted motion.

Step 2: Read why the planner declined

The planner made a correct decision with the wrong input. It compared the 31,457,280-byte estimate for stores to the 10 MB threshold, found it too big to broadcast, and chose the safe strategy. Nothing is misconfigured. The default simply does not fit this table.

Step 3: Keep the grain, move the small side

The rows stay order-shaped: lead from retail_orders, look up each store_id, keep order_id and amount. Then wrap stores in F.broadcast(...). Each executor gets one copy of 5,000 rows and builds a hash table from it, so every orders partition probes it where it already sits, with no Exchange and no Sort.

Step 4: Prove it in the plan

Run explain() again, or open the SQL tab. The fix is real only when BroadcastHashJoin with BuildRight replaces SortMergeJoin and the orders scan feeds the probe with no Exchange above it.

What the job was running
orders = spark.table("retail_orders")
stores = spark.table("stores")

result = (
    orders
    .join(stores, on="store_id", how="inner")
    .select("order_id", "amount", "store_name", "region")
)

Correct rows, and the planner shuffles both sides because stores is over the threshold.

Broadcast the dimension
orders = spark.table("retail_orders")
stores = spark.table("stores")

# stores is the small side: ship a copy to every executor so orders never moves
result = (
    orders
    .join(F.broadcast(stores), on="store_id", how="inner")
    .select("order_id", "amount", "store_name", "region")
)

Same rows, and the plan no longer shuffles the 80 GB side.

Before: SortMergeJoin
Two Exchange hashpartitioning(store_id, 200) nodes and two Sorts. 81.2 GB shuffle read, 4.8 s of GC on the hot task, 12m 04s.
After: BroadcastHashJoin
One BroadcastExchange on stores, none on orders. 0 MB shuffle read on the orders task, 380 ms of GC, 1m 58s.

Common follow-up questions