Back to Apache Spark: Core
Q11Question 11 of 72

How to verify Spark actually used a Broadcast Join

Key Concepts
  • Call df.explain('formatted') and look for BroadcastExchange on the small side feeding a BroadcastHashJoin node
  • If you see two Exchange hashpartitioning nodes feeding SortMergeJoin, the broadcast did not happen
  • In the Spark UI Stages tab, the large side should show zero shuffle read/write for a confirmed broadcast join
  • The SQL tab shows the plan as a diagram with the broadcast data size annotated on the exchange node
  • With AQE enabled, the static explain may differ from the final executed plan: trust the Spark UI
  • A BroadcastExchange with a very large size is a warning that you broadcast something too big

Definition

Confirm a broadcast join by seeing BroadcastExchange plus BroadcastHashJoin in explain output and no shuffle on the large side in the Spark UI.
  • Physical plan – call df.explain("formatted") and read bottom-up; a broadcast join shows BroadcastExchange on the small side and BroadcastHashJoin as the join node.
  • Negative signal – two Exchange hashpartitioning nodes feeding SortMergeJoin means the broadcast did not happen.
  • Spark UI – the SQL tab shows the plan diagram with broadcast data size; the Stages tab shows zero shuffle on the large side.
  • AQE caveat – with AQE, the final executed plan in the Spark UI may differ from the static explain; trust the UI.

Claiming the National Bank photocopied the small list for every clerk is one thing; proving it is another. You check the work order (the explain plan) for the instruction "photocopy and distribute" next to "each clerk matches locally." Then you check the activity log (the Spark UI) and confirm the big pile was never carted across the room.

Understanding the Concepts

How do you check the physical plan?

Call explain("formatted") and read the plan bottom-up because execution flows from leaves to root. For a broadcast join, you need two things: a BroadcastExchange node on the small side and a BroadcastHashJoin node as the join operator.

PYTHON
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.functions import broadcast

spark = SparkSession.builder.appName("NBI-VerifyBroadcast").getOrCreate()

txns = spark.read.parquet("s3://nbi-data-lake/transactions/")
branches = spark.read.parquet("s3://nbi-data-lake/branches/")

result = txns.join(broadcast(branches), "branchId")
result.explain("formatted")
TEXT
== Physical Plan ==
*(1) BroadcastHashJoin [branchId], [branchId], Inner, BuildRight
:- *(1) FileScan parquet [branchId,amount,...]
+- BroadcastExchange HashedRelationBroadcastMode
   +- *(2) FileScan parquet [branchId,branchName,...]

What does a failed broadcast look like?

If the broadcast did not happen, you see two Exchange hashpartitioning nodes (shuffles) feeding a SortMergeJoin.

PYTHON
# Suppose the broadcast hint was ignored or not applied
result_bad = txns.join(branches, "branchId")
result_bad.explain("formatted")
DIAGRAM
== Physical Plan ==
*(3) SortMergeJoin [branchId], [branchId], Inner
:- *(1) Sort [branchId ASC]
:  +- Exchange hashpartitioning(branchId, 200)    ← SHUFFLE!
:     +- FileScan parquet [branchId,amount,...]
+- *(2) Sort [branchId ASC]
   +- Exchange hashpartitioning(branchId, 200)    ← SHUFFLE!
      +- FileScan parquet [branchId,branchName,...]

How do you use the Spark UI SQL tab?

After the job runs, the SQL tab shows the physical plan as a diagram with metrics. The BroadcastExchange node is annotated with the broadcast data size.

PYTHON
# Run the query to populate the Spark UI
result = txns.join(broadcast(branches), "branchId")
result.write.mode("overwrite").parquet("s3://nbi-data-lake/enriched/")

# Navigate to: http://driver-host:4040/SQL/
# Click the query → see the diagram
#
# Look for:
# - BroadcastExchange node with "data size: 4.2 KB" (small!)
# - BroadcastHashJoin node (correct join type)
#
# Warning signal:
# - BroadcastExchange with "data size: 800 MB" → too big!

How do you use the Stages tab?

The Stages tab confirms that the large side was not shuffled. For a broadcast join, the large side should show zero shuffle read bytes and zero shuffle write bytes.

PYTHON
# Navigate to: http://driver-host:4040/stages/
#
# For the join stage:
# - Shuffle Read: 0 B (large side NOT shuffled)
# - Shuffle Write: 0 B
# - Broadcast size: 4.2 KB (small side broadcast)
#
# If you see large Shuffle Read/Write on BOTH sides,
# that is a SortMergeJoin, not a BroadcastHashJoin.

How does AQE affect verification?

With AQE enabled, the initial explain output is the plan BEFORE AQE optimizations. AQE may change the plan at runtime, for example switching a planned SortMergeJoin to a BroadcastHashJoin. The Spark UI SQL tab shows the FINAL plan after AQE adjustments.

PYTHON
spark.conf.set("spark.sql.adaptive.enabled", "true")

# Static explain: may show SortMergeJoin
result.explain("formatted")  # Pre-AQE plan

# Spark UI SQL tab: shows BroadcastHashJoin (post-AQE)
# TRUST THE UI for the final executed plan

# Look for "AdaptiveSparkPlan isFinalPlan=true" in the UI

How do you use explain with cost mode?

The cost mode shows the estimated sizes that drove the optimizer's decision, which helps diagnose why a broadcast did or did not happen.

PYTHON
# Extended explain shows logical + physical plan
result.explain("extended")

# Cost explain shows estimated sizes
result.explain("cost")

# If the small side's estimate is above the threshold,
# Spark planned SortMergeJoin instead of BroadcastHashJoin.

What are the warning signs to look for?

Even if a broadcast join succeeded, there are signals that something may be wrong.

PYTHON
# Warning 1: BroadcastExchange with large data size
# If the broadcast size is 500 MB, you broadcast too much.
# Risk: driver OOM or executor memory pressure.

# Warning 2: Broadcast join but slow tasks
# Building the hash table takes time for large broadcasts.
# Check task durations in the Stages tab.

# Warning 3: AQE-adjusted plan differs from hint
# You hinted broadcast, but AQE shows SortMergeJoin.
# The table was too large even after AQE checked.

What is the complete verification checklist?

A systematic approach to confirming a broadcast join.

PYTHON
# Step 1: Check the static plan
result.explain("formatted")
# ✓ BroadcastExchange on small side
# ✓ BroadcastHashJoin as join node

# Step 2: Run the query
result.write.parquet("s3://nbi-data-lake/output/")

# Step 3: Check Spark UI SQL tab
# ✓ Final plan shows BroadcastHashJoin
# ✓ Broadcast data size is small (KB or low MB)

# Step 4: Check Spark UI Stages tab
# ✓ Large side: 0 shuffle read/write
# ✓ No Sort stage for the join

# Step 5: Check task durations
# ✓ Join tasks complete quickly
# ✓ No stragglers building a huge hash table

What should a senior engineer know?

A senior engineer always verifies critical joins rather than assuming hints worked. They know that the Spark UI is the source of truth for AQE queries, and they check broadcast data sizes to catch accidental large broadcasts.

PYTHON
# Senior practice: verify every critical join
result = txns.join(broadcast(branches), "branchId")
result.explain("formatted")
# If BroadcastHashJoin → proceed
# If SortMergeJoin → investigate why hint was ignored

# After execution: check Spark UI for actual broadcast size
# and zero shuffle on the large side

Visual Overview

DIAGRAM
  Verification flow:

  Step 1: explain("formatted")
  ┌──────────────────────────────────────────────┐
  │ BroadcastHashJoin [branchId] ← JOIN NODE     │
  │ :- FileScan parquet (txns)   ← large side    │
  │ +- BroadcastExchange         ← BROADCAST!    │
  │    +- FileScan parquet (branches) ← small side│
  └──────────────────────────────────────────────┘
  ✓ Good: BroadcastExchange + BroadcastHashJoin

  Step 2: Spark UI Stages tab
  ┌──────────────────────────────────────┐
  │ Shuffle Read:  0 B  (large side)    │ ← No shuffle!
  │ Shuffle Write: 0 B  (large side)    │
  │ Broadcast:     4.2 KB (small side)  │
  └──────────────────────────────────────┘
  ✓ Good: zero shuffle on large side

Key Takeaway

Verifying a broadcast join requires two independent checks: the physical plan and the Spark UI. In the plan, you call explain("formatted") and read bottom-up. A real broadcast join shows a BroadcastExchange node on the small side feeding a BroadcastHashJoin node. If you see two Exchange hashpartitioning nodes feeding a SortMergeJoin, the broadcast did not happen.

In the Spark UI, the SQL tab shows the plan diagram with the broadcast data size annotated. The Stages tab confirms that the large side has zero shuffle read and write bytes. If both sides show large shuffle numbers, it is a sort-merge join.

With AQE enabled, the static explain output may differ from what actually ran, because AQE can switch plans at runtime. The Spark UI's final executed plan is the source of truth.

At the National Bank of India, I used this exact process when a hinted join was slow. The explain plan revealed SortMergeJoin instead of BroadcastHashJoin, which told me the hint was not applied. After fixing the issue, the plan showed BroadcastExchange and BroadcastHashJoin, and the Stages tab confirmed zero shuffle on the transactions side.

Remember one sentence: verify a broadcast join by checking for BroadcastExchange plus BroadcastHashJoin in the physical plan and zero shuffle on the large side in the Spark UI.

---

Share
Use ← → arrow keys to navigate