1. The Problem
Joining a 50 GB tasks fact DataFrame with a 15 MB tenant dimension table triggered a full cluster Sort Merge Join shuffle, consuming 18 GB of network transfer.
2. What I Initially Thought
3. What I Learned
Wrapping small DataFrames in `broadcast(dim_df)` copies the dimension table to every executor once, converting a wide Sort Merge Join into a fast local Broadcast Hash Join with 0 shuffle.
4. What I Built
Refactored PySpark join pipeline explicitly broadcasting small dimension tables (`final_df = fact_df.join(broadcast(dim_df), 'tenant_id')`).
from pyspark.sql.functions import broadcast
# Copy small 15MB dimension table to all executors -> ZERO SHUFFLE!
joined_df = fact_tasks.join(
broadcast(dim_tenants),
on="tenant_id",
how="inner"
)5. The Experiment
Joining 50 GB fact table with 15 MB dimension table via default Sort Merge Join. Network Shuffle: 18 GB. Duration: 3.5 minutes.
Wrapped dimension table in `broadcast()` hint, converting join type to Broadcast Hash Join.
Network Shuffle Read bytes reduced to 0 B; execution duration dropped from 3.5 minutes to 28 seconds.
6. What Went Wrong
Attempted to broadcast a 3 GB DataFrame, which exceeded Driver JVM memory limits and caused a Driver OutOfMemoryError.
7. Engineering Decision & Trade-offs
Enforced strict policy: Only use `broadcast()` for dimension tables smaller than 100 MB to prevent Driver OOM crashes.
8. What I Would Do Differently in Production
Ensure Driver node memory is sized sufficiently when increasing `spark.sql.autoBroadcastJoinThreshold` in enterprise clusters.
Questions I Can Now Answer Confidently in an Interview:
- How does a Broadcast Hash Join differ from a Sort Merge Join in Apache Spark?
- What are the risks of using broadcast() on a DataFrame that exceeds Driver memory?
- How do you inspect the Spark Physical Plan to confirm a Broadcast Hash Join is executing?