CASE STUDIES LIST/ HOME
← Case #14#15 / 25Case #16 →
CASE STUDY #15Phase 4: Spark & Distributed ProcessingCLASSIFICATION: EXPERIMENT

Replacing Network Shuffle With Broadcast Joins

Eliminating Shuffle in PySpark Using Broadcast Hash Joins

Apache SparkPySparkSpark UI

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

"I assumed Spark would automatically optimize joins with small lookup tables. But default `autoBroadcastJoinThreshold` was set below the dimension table size."

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.

Broadcast Hash JoinSort Merge JoinDriver Memory LimitsDimension Table Lookups

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

BEFORE

Joining 50 GB fact table with 15 MB dimension table via default Sort Merge Join. Network Shuffle: 18 GB. Duration: 3.5 minutes.

CHANGE APPLIED

Wrapped dimension table in `broadcast()` hint, converting join type to Broadcast Hash Join.

AFTER RESULT

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?

Expected / Verified Evidence

•Spark UI DAG visualization showing zero Shuffle Read on Stage 2
•Spark Physical Plan `BroadcastHashJoin` output log
•PySpark benchmark script (pipelines/gold/broadcast_join.py)
BACK TO ALL CASE STUDIESNEXT: CASE #16 (Most Workers Finished Fast. The Last Worker Took Forever.)