Definition
There are three situations where broadcast joins should be avoided.
- Small side too large – The smaller table's deserialized in-memory size exceeds executor memory, causing driver or executor OOM. This is far worse than a slow sort-merge join.
- Both sides large – Neither side qualifies for broadcasting. The correct strategy is a shuffle sort-merge join.
- Tiny inputs – When both inputs are so small that the fixed overhead of collecting and broadcasting is not worth it.
- Key rule – 'Smaller does not mean safe to broadcast' captures that broadcasting cares about the absolute size against executor memory, not about which side is smaller.
Always ask: does this side's deserialized form fit comfortably in executor memory with room to spare?
At the National Bank of India, photocopying the smaller list for every clerk only works if that list is genuinely small: a few pages. If the 'smaller' list is still ten thousand pages, photocopying it for every one of the hundred clerks means a million pages, and the copy room collapses. Being smaller than the other list does not make it small enough to copy everywhere; it just makes it less huge.
Understanding the Concepts
When is the small side too large?
Broadcasting materializes the small side on the driver and sends a full copy to every executor, where it is deserialized into an in-memory hash table. The deserialized form can be several times the on-disk size. A table of a few hundred megabytes on disk can balloon to gigabytes in memory.
# DANGEROUS: the accounts table is 800 MB on disk
# After deserialization, it becomes ~3 GB in memory
# Broadcasting this crashes executors with OOM
from pyspark.sql.functions import broadcast
txns = spark.read.parquet("/data/bank/transactions") # 500 GB
accounts = spark.read.parquet("/data/bank/accounts") # 800 MB on disk
# This will likely cause OutOfMemoryError
result = txns.join(broadcast(accounts), "account_id")
# ERROR: java.lang.OutOfMemoryError: Java heap space
# The job CRASHES instead of running slowly
On-disk size vs in-memory size:
Parquet (compressed, columnar): 800 MB
Deserialized Java objects: ~3 GB (4x expansion)
Broadcast to 50 executors: 50 x 3 GB hash table overhead
Each executor may run 4 tasks: 4 x reference to broadcast
Total memory pressure: significant
When are both sides large?
When both tables are large, neither qualifies for broadcasting. Forcing a broadcast with a hint is a classic mistake.
# WRONG: both sides are large, neither can be broadcast
txns = spark.read.parquet("/data/bank/transactions") # 500 GB
loans = spark.read.parquet("/data/bank/loan_payments") # 200 GB
# This hint forces a broadcast of 200 GB → guaranteed OOM
result = txns.join(broadcast(loans), "account_id") # DO NOT DO THIS
# CORRECT: let Spark use sort-merge join
result = txns.join(loans, "account_id")
result.explain()
# Physical plan: SortMergeJoin (safe, memory-efficient)
Why does 'smaller does not mean safe to broadcast'?
People conflate relative size (smaller of two) with absolute size (fits in memory). Broadcasting cares only about the absolute size.
# Scenario: transactions is 500 GB, accounts is 8 GB
# Accounts is SMALLER than transactions
# But 8 GB does NOT fit in executor memory (typically 4-8 GB)
# The question is NOT: "is this the smaller side?"
# The question IS: "does this side's deserialized form fit
# comfortably in executor memory with room for the rest?"
accounts_size_gb = 8
executor_memory_gb = 8
task_overhead_gb = 2 # GC, shuffle buffers, etc.
available_for_broadcast = executor_memory_gb - task_overhead_gb
print(f"Available: {available_for_broadcast} GB") # 6 GB
print(f"Accounts deserialized: ~{accounts_size_gb * 3} GB") # 24 GB
print("UNSAFE to broadcast!") # 24 GB >> 6 GB
Why is raising autoBroadcastJoinThreshold dangerous?
The auto-broadcast threshold tells Spark the maximum estimated table size it should broadcast automatically. Raising it globally affects every join in every query.
# Default: 10 MB (conservative, safe)
spark.conf.get("spark.sql.autoBroadcastJoinThreshold") # 10485760 (10 MB)
# DANGEROUS: raising to 1 GB
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "1073741824")
# Now Spark will try to broadcast ANY table up to 1 GB
# across ALL joins in ALL queries
# One join with a 900 MB table → OOM on every executor
# SAFE: use per-query broadcast hints for known-small tables
result = txns.join(broadcast(branches), "branch_code") # branches = 5 MB
How do you determine if a table is safe to broadcast?
Check the actual deserialized size, not just the on-disk size. Account for executor memory and concurrent tasks.
# Check actual sizes before deciding
accounts = spark.read.parquet("/data/bank/accounts")
# On-disk size
import os
path = "/data/bank/accounts"
total_size = sum(f.stat().st_size for f in os.scandir(path) if f.is_file())
print(f"On-disk: {total_size / (1024**2):.0f} MB")
# Rough in-memory estimate (typically 2-4x on-disk for Parquet)
estimated_memory = total_size * 3
print(f"Estimated in-memory: {estimated_memory / (1024**2):.0f} MB")
# Compare against executor memory
executor_memory_mb = 8192 # 8 GB
if estimated_memory / (1024**2) < executor_memory_mb * 0.3: # 30% budget
print("Safe to broadcast")
else:
print("NOT safe to broadcast - use sort-merge join")
What is the correct join strategy for each scenario?
| Scenario | Strategy | Example |
|---|---|---|
| One side < 10 MB | Auto broadcast | txns JOIN branches (5 MB) |
| One side 10 MB - 100 MB | Broadcast hint (verify first) | txns JOIN products (50 MB) |
| One side > 100 MB | Sort-merge join | txns JOIN accounts (8 GB) |
| Both sides large | Sort-merge join | txns JOIN loans (200 GB) |
| Post-filter makes one side small | Let AQE handle it | Filtered dimension |
# Let AQE handle post-filter cases automatically
# AQE measures the real size after filtering and switches if small enough
spark.conf.set("spark.sql.adaptive.enabled", "true")
# AQE may switch this to broadcast at runtime
result = txns.join(accounts.filter("status = 'ACTIVE'"), "account_id")
What about executor memory and concurrent tasks?
Many tasks on the same executor each reference the broadcast. While broadcast data is shared (not copied per task), the hash table build and memory pressure still apply.
Executor with 8 GB memory, 4 concurrent tasks: Broadcast table: 2 GB deserialized (shared) Each task needs: ~1 GB for processing Total: 2 GB broadcast + 4 x 1 GB tasks = 6 GB Remaining for GC/overhead: 2 GB (tight!) If broadcast table were 4 GB: Total: 4 GB + 4 GB = 8 GB → OOM
What should a senior engineer know?
- OOM from broadcasting is worse than a slow sort-merge join: the job crashes instead of merely running slowly.
- Use broadcast hints sparingly and only for known-small tables. Let AQE handle post-filter cases.
- Monitor the Spark UI for broadcast size warnings in the SQL plan.
- The 30% rule: a broadcast table should use no more than 30% of executor memory to leave room for tasks and GC.
# Safe pattern: broadcast only genuinely tiny dimensions
result = txns.join(broadcast(branches), "branch_code") # 5 MB: safe
result = txns.join(broadcast(currencies), "currency_code") # 1 MB: safe
result = txns.join(accounts, "account_id") # 8 GB: sort-merge
Visual Overview
Broadcast Join Decision:
Is the small side < 30% of executor memory (deserialized)?
│
├── YES → Broadcast Join (fast, no shuffle of large side)
│ ┌──────────┐
│ │ Small │──► copied to every executor
│ │ table │ as hash table
│ └──────────┘
│
└── NO → Sort-Merge Join (safe, memory-efficient)
┌──────────┐ ┌──────────┐
│ Table A │ │ Table B │
│ shuffle │ │ shuffle │
│ + sort │ │ + sort │
└─────┬────┘ └─────┬────┘
└─── merge ───┘
"Smaller" ≠ "Safe to broadcast"
┌───────────────────────────────────────┐
│ txns: 500 GB accounts: 8 GB │
│ accounts is SMALLER but 8 GB is NOT │
│ small enough to broadcast safely │
└───────────────────────────────────────┘
Key Takeaway
The most dangerous broadcast join mistake is broadcasting a 'small' side that is not actually small enough to fit in executor memory. Broadcasting materializes the table on the driver and copies it to every executor, where it becomes a hash table that can be several times its on-disk size. If it does not fit, the job crashes with an OOM error, which is far worse than a slow sort-merge join.
The production rule 'smaller does not mean safe to broadcast' captures that broadcasting cares about the absolute size against executor memory, not about which side happens to be smaller. A table can be the smaller of two and still be many gigabytes. The right question is: does this side's deserialized form fit comfortably in executor memory with room for the rest?
At the National Bank of India, a colleague saw that accounts was smaller than transactions and added a broadcast hint: accounts was 8 GB, and broadcasting it crashed executors. We removed the hint, used a sort-merge join, and only broadcast genuinely tiny dimensions like branches (5 MB). Raising the autoBroadcastJoinThreshold globally is equally dangerous because it tells Spark to broadcast larger sizes across every join.
The safe pattern is: broadcast only genuinely small tables, let AQE handle post-filter cases, and never raise the global threshold recklessly.
Remember one sentence: being the smaller side does not make a table safe to broadcast; only fitting comfortably in executor memory does, so use sort-merge join for anything larger.
---