Definition
- Broadcast Hash Join (BHJ) – broadcasts the small side to all executors, builds a hash table, probes with the large side; no shuffle of the large table.
- Shuffle Sort Merge Join (SSMJ) – shuffles both sides by join key, sorts each partition, merges; the robust default for two large tables.
- Shuffle Hash Join (SHJ) – shuffles both sides, builds a hash table on the smaller side per partition; avoids the sort cost.
- Broadcast Nested Loop Join (BNLJ) – broadcasts one side and compares every pair; used for non-equi joins where no hash or sort key exists.
- Selection order – BHJ if a side is under the threshold, then SHJ if conditions favor it, then SSMJ for large equality joins, then BNLJ for non-equi.
The National Bank of India matches two lists. BHJ: photocopy the tiny branch list for every clerk so each matches locally. SSMJ: sort both huge lists by account number and zip them together. SHJ: group both lists into buckets and build a lookup in each bucket. BNLJ: compare every pair for fuzzy matches like "amount within ten percent" where there is no exact key.
Understanding the Concepts
What is Broadcast Hash Join (BHJ)?
BHJ is the fastest join when one side is small. Spark collects the small table to the Driver, broadcasts it to every executor, and each executor builds an in-memory hash table. The large side's local partitions are streamed and probed against that hash table. The large side is never shuffled.
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.functions import broadcast
spark = SparkSession.builder.appName("NBI-JoinStrategies").getOrCreate()
txns = spark.read.parquet("s3://nbi-data-lake/transactions/") # 500M rows
branches = spark.read.parquet("s3://nbi-data-lake/branches/") # 200 rows
# BHJ: broadcast the tiny branch table
result = txns.join(broadcast(branches), "branchId")
result.explain("formatted")
== Physical Plan ==
*(1) BroadcastHashJoin [branchId], [branchId], Inner, BuildRight
:- FileScan parquet [branchId,amount,...]
+- BroadcastExchange HashedRelationBroadcastMode
+- FileScan parquet [branchId,branchName,...]
What is Shuffle Sort Merge Join (SSMJ)?
SSMJ is the default for two large tables. Spark shuffles both sides by the join key so matching keys land in the same partition, sorts each partition, and then merges the two sorted streams. It is robust because merging is a streaming operation that works even with data too large for memory.
accounts = spark.read.parquet("s3://nbi-data-lake/accounts/") # 100M rows
# SSMJ: both sides are large
result = txns.join(accounts, "accountId")
result.explain("formatted")
== Physical Plan ==
*(3) SortMergeJoin [accountId], [accountId], Inner
:- *(1) Sort [accountId ASC]
: +- Exchange hashpartitioning(accountId, 200)
: +- FileScan parquet [accountId,amount,...]
+- *(2) Sort [accountId ASC]
+- Exchange hashpartitioning(accountId, 200)
+- FileScan parquet [accountId,customerName,...]
What is Shuffle Hash Join (SHJ)?
SHJ shuffles both sides by the join key but skips sorting. Instead, it builds a hash table on the smaller side within each partition and probes it with the other side. It can be faster than SSMJ when the smaller side fits in memory per partition, but risks out-of-memory if it does not.
# Force shuffle hash join for demonstration
spark.conf.set("spark.sql.join.preferSortMergeJoin", "false")
result = txns.join(accounts, "accountId")
result.explain("formatted")
# May show ShuffledHashJoin instead of SortMergeJoin
What is Broadcast Nested Loop Join (BNLJ)?
BNLJ is the fallback for joins with no equality condition, like matching amounts within a tolerance. There is no key to hash or sort, so Spark broadcasts one side and loops over every pair. This is O(N * M) and very slow.
# Non-equi join: amount within 10% of a price band
price_bands = spark.createDataFrame(
[("LOW", 0, 1000), ("MED", 1000, 10000), ("HIGH", 10000, 1000000)],
["band", "lower", "upper"]
)
# Range condition: no equality key → BNLJ
result = txns.join(
broadcast(price_bands),
(F.col("amount") >= F.col("lower")) & (F.col("amount") < F.col("upper"))
)
result.explain("formatted")
== Physical Plan ==
BroadcastNestedLoopJoin BuildRight, Inner, ...
How does the optimizer choose?
The optimizer evaluates join strategies in a priority order based on size estimates, the join condition, and the join type.
| Priority | Condition | Strategy |
|---|---|---|
| 1 | Equality join + one side under broadcast threshold | Broadcast Hash Join |
| 2 | Equality join + one side fits per partition | Shuffle Hash Join |
| 3 | Equality join + two large sides | Shuffle Sort Merge Join |
| 4 | Non-equality join | Broadcast Nested Loop Join |
# Check the broadcast threshold (default 10 MB)
print(spark.conf.get("spark.sql.autoBroadcastJoinThreshold")) # 10485760
# If branches estimated at 5 MB → BHJ automatically
# If branches estimated at 50 MB → SSMJ (unless hinted)
How do bad statistics cause wrong join strategies?
Size estimates drive the optimizer's decision. Stale or missing statistics can cause Spark to skip a broadcast it should have done, or attempt one that blows up memory.
# Problem: stale statistics say customers table is 500 MB
# Reality: after filtering to VIP, it is only 2 KB
vip = spark.read.parquet("s3://nbi-data-lake/customers/") \
.filter(F.col("tier") == "VIP")
# Without hint: Spark plans SSMJ (wrong, wastes shuffle)
result_slow = txns.join(vip, "customerId")
# With hint: force broadcast (correct, skips shuffle)
result_fast = txns.join(broadcast(vip), "customerId")
result_fast.explain("formatted")
How does AQE change join strategies at runtime?
AQE can convert a planned SSMJ to BHJ at runtime when it discovers a side is actually small after earlier stages complete. This fixes wrong estimates without hints.
spark.conf.set("spark.sql.adaptive.enabled", "true")
# Catalyst plans SSMJ (bad estimate). AQE sees VIP is tiny → BHJ.
result = txns.join(vip, "customerId")
result.write.parquet("s3://nbi-data-lake/vip-txns/")
# Check Spark UI SQL tab for the FINAL executed plan:
# Should show BroadcastHashJoin, not SortMergeJoin
How do you verify which strategy was used?
Always check the physical plan with explain and verify in the Spark UI after execution.
# Before execution: static plan
result.explain("formatted")
# Look for: BroadcastExchange + BroadcastHashJoin (BHJ)
# or: Exchange + SortMergeJoin (SSMJ)
# or: ShuffledHashJoin (SHJ)
# or: BroadcastNestedLoopJoin (BNLJ)
# After execution with AQE: check Spark UI SQL tab
# The final plan may differ from the static explain
What should a senior engineer know?
A senior engineer keeps statistics fresh, uses broadcast hints when estimates are unreliable, and reads the physical plan to confirm the expected strategy. They also know that the join type constrains which side can be broadcast (you cannot broadcast the left side of a left outer join).
# Senior checklist for joins:
# 1. Is one side small? → Force or verify BHJ
# 2. Both sides large? → SSMJ (default, robust)
# 3. Non-equi condition? → BNLJ (keep inputs small!)
# 4. AQE enabled? → Check final plan in Spark UI
# 5. Statistics fresh? → ANALYZE TABLE for Hive tables
spark.sql("ANALYZE TABLE nbi.customers COMPUTE STATISTICS")
Visual Overview
Broadcast Hash Join (BHJ):
Small (200 rows) Large (500M rows)
│ │
▼ broadcast │ no shuffle!
┌──────────┐ ┌──────────┐
│ Hash map │──probe──────→│ Executor │
└──────────┘ └──────────┘
Shuffle Sort Merge Join (SSMJ):
Large A Large B
│ │
▼ shuffle by key ▼ shuffle by key
▼ sort ▼ sort
┌──────────┐ ┌──────────┐
│ Sorted A │──merge──────→│ Sorted B │
└──────────┘ └──────────┘
Optimizer priority:
1. BHJ (small side < threshold)
2. SHJ (one side fits per partition)
3. SSMJ (two large, equality)
4. BNLJ (non-equi, last resort)
Key Takeaway
Spark has four physical join strategies. Broadcast Hash Join broadcasts the small side to every executor and probes it with the large side, avoiding any shuffle of the large table. Shuffle Sort Merge Join shuffles both sides by join key, sorts each partition, and merges, which is robust and the default for two large tables. Shuffle Hash Join shuffles and builds a hash table on the smaller side per partition, avoiding the sort cost. Broadcast Nested Loop Join broadcasts one side and compares every pair, used only for non-equi joins.
The optimizer chooses in priority order: BHJ if a side is under the broadcast threshold, SHJ if conditions favor it, SSMJ for large equality joins, and BNLJ for non-equi joins. AQE can switch from SSMJ to BHJ at runtime when actual sizes turn out small.
At the National Bank of India, joining transactions to the tiny branch dimension was always BHJ with no shuffle of the big table. Joining transactions to the large accounts table was SSMJ. A risk report matching trades to price bands within a tolerance fell to BNLJ. The nuance is that size estimates drive the decision, so stale statistics cause wrong plans, which is why I keep statistics fresh and use broadcast hints when needed.
Remember one sentence: broadcast for a small side, sort-merge for two large sides, and the optimizer picks by size, threshold, and join condition.
---