Back to Apache Spark: Core
Q9Question 9 of 72

Broadcast Join: sending the small dataset to every executor

Key Concepts
  • A Broadcast Join copies the small table to every executor so the large table joins locally with no shuffle
  • Each executor builds an in-memory hash table from the broadcast and probes it with local partitions of the large side
  • In PySpark: from pyspark.sql.functions import broadcast; big.join(broadcast(small), 'key')
  • Spark also broadcasts automatically when a side's estimated size is below the autoBroadcastJoinThreshold (10 MB)
  • The small side must fit in executor memory after deserialization, which can be several times its on-disk size
  • Broadcast joins support inner joins and most outer joins but you cannot broadcast the outer side of an outer join

Definition

A Broadcast Join copies the small table to every executor so the large table joins locally with no shuffle: in PySpark, wrap the small side in broadcast(small).
  • Mechanism – Spark collects the small table to the Driver, broadcasts it to all executors, each builds an in-memory hash table, and probes it with local partitions of the large side.
  • Benefit – the large table is never shuffled, eliminating the most expensive step of a join.
  • PySpark syntax – from pyspark.sql.functions import broadcast; big.join(broadcast(small), "key")
  • Auto-broadcast – Spark broadcasts automatically when a side's estimated size is under spark.sql.autoBroadcastJoinThreshold (default 10 MB).
  • Constraint – the small side must fit in executor memory; you cannot broadcast the outer side of an outer join.

The National Bank of India needs to tag every transaction with its branch name. There are only 50 branches but a billion transactions. Instead of carting a billion transactions to a central desk, the bank photocopies the tiny 50-row branch list and hands one copy to every clerk. Each clerk looks up branch names for their own transactions locally, with no need to move the big pile.

Understanding the Concepts

What is a Broadcast Join?

A Broadcast Join (Broadcast Hash Join) sends the entire small table to every executor. Each executor builds an in-memory hash table from it and streams the large table's local partitions, probing the hash table for matches. The large side never moves across the network.

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

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

txns = spark.read.parquet("s3://nbi-data-lake/transactions/")   # 500M rows
branches = spark.read.parquet("s3://nbi-data-lake/branches/")   # 50 rows

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

How does it work step by step?

The mechanism has three steps. First, Spark collects the small table to the Driver and broadcasts it to all executors. Second, each executor builds a hash table keyed by the join column. Third, each executor streams its local partitions of the large table and probes the hash table for matches.

PYTHON
# Step 1: Driver collects branches (50 rows, ~2 KB)
# Step 2: Broadcast to all 30 executors (each gets a copy)
# Step 3: Each executor builds hash map: {branchId: branchRow}
# Step 4: Each executor streams its txn partitions and probes

# No shuffle of the 500M-row transactions table!
# Each partition stays where it is.

What is the PySpark syntax?

Import the broadcast function and wrap the small DataFrame.

PYTHON
# PySpark
from pyspark.sql.functions import broadcast
result = txns.join(broadcast(branches), "branchId")

# Scala / Java equivalent:
# val result = txns.join(broadcast(branches), "branchId")

# Spark SQL equivalent:
# SELECT /*+ BROADCAST(b) */ t.*, b.branchName
# FROM transactions t
# JOIN branches b ON t.branchId = b.branchId

When does Spark broadcast automatically?

Spark broadcasts a side automatically when its estimated serialized size is at or below spark.sql.autoBroadcastJoinThreshold (default 10 MB). No hint is needed.

PYTHON
# Check the threshold
print(spark.conf.get("spark.sql.autoBroadcastJoinThreshold"))  # 10485760

# branches table is ~2 KB: well under 10 MB → auto-broadcast
result = txns.join(branches, "branchId")  # No broadcast() call needed
result.explain("formatted")
# Shows BroadcastHashJoin because branches < threshold

When should you use the explicit broadcast hint?

Use the explicit broadcast() when the auto-broadcast misses because of stale or missing statistics.

PYTHON
# Case 1: Stale statistics overestimate the table size
customers = spark.read.parquet("s3://nbi-data-lake/customers/")
vip = customers.filter(F.col("tier") == "VIP")  # Only 5000 rows

# Spark thinks customers is 500 MB, plans sort-merge.
# With hint: forces broadcast of the tiny filtered result.
result = txns.join(broadcast(vip), "customerId")

# Case 2: Table is just above the 10 MB threshold
products = spark.read.parquet("s3://nbi-data-lake/products/")  # 12 MB
# Without hint: sort-merge (12 MB > 10 MB threshold)
# With hint: broadcast (12 MB fits in executor memory)
result = txns.join(broadcast(products), "productId")

What are the limits and costs of broadcast joins?

Broadcasting is not free. The small side is materialized on the Driver and copied to every executor. After deserialization, it can be several times its on-disk size.

PYTHON
# Cost: Driver must fit the small table in memory
# Cost: Each executor gets a copy (N executors × table size)
# Risk: Broadcasting a 500 MB table to 30 executors = 15 GB total

# WRONG: broadcasting a large table
result = txns.join(broadcast(accounts), "accountId")  # 100M rows!
# Driver OOM or executor OOM

# RIGHT: only broadcast genuinely small tables
result = txns.join(broadcast(branches), "branchId")   # 50 rows

What join types support broadcast?

Broadcast Hash Join supports inner joins and most outer joins, but you cannot broadcast the outer side.

PYTHON
# Inner join: broadcast either side
result = txns.join(broadcast(branches), "branchId", "inner")  # OK

# Left outer: broadcast the RIGHT side only
result = txns.join(broadcast(branches), "branchId", "left")   # OK

# Right outer: broadcast the LEFT side only
result = branches.join(broadcast(txns), "branchId", "right")  # WRONG!
# Cannot broadcast the outer (right) side of a right join

# Full outer: broadcast not supported
result = txns.join(broadcast(branches), "branchId", "full")   # Ignored

How do you verify a broadcast join was used?

Check the physical plan for BroadcastExchange and BroadcastHashJoin. In the Spark UI, the large side should show no shuffle reads.

PYTHON
result = txns.join(broadcast(branches), "branchId")
result.explain("formatted")

# Look for:
# BroadcastExchange → small side broadcast
# BroadcastHashJoin → join using the broadcast

# Spark UI → Stages tab:
# Large side: 0 shuffle read bytes
# Small side: small broadcast size

What should a senior engineer know?

A senior engineer verifies broadcast joins by checking both the plan and the Spark UI. With AQE, the final executed plan may differ from the static explain. They also know that deserialized size in memory can be 2x to 10x the on-disk Parquet size, so the 10 MB threshold is conservative.

PYTHON
# Senior practice:
# 1. Use broadcast() for known-small sides
# 2. Verify with explain() for BroadcastExchange + BroadcastHashJoin
# 3. Check Spark UI for actual broadcast size
# 4. With AQE, trust the UI's final plan over static explain
# 5. Never broadcast a side larger than executor memory

Visual Overview

DIAGRAM
  Broadcast Join mechanism:

  Small table (50 rows)         Large table (500M rows)
       │                              │
       ▼ collect to Driver            │ stays on executors
  ┌──────────┐                        │
  │  Driver   │                       │
  │  50 rows  │                       │
  └────┬──────┘                       │
       │ broadcast to all executors   │
       ▼                              ▼
  ┌──────────┐  ┌──────────┐  ┌──────────┐
  │Executor 1│  │Executor 2│  │Executor 3│
  │ Hash map │  │ Hash map │  │ Hash map │
  │ + probe  │  │ + probe  │  │ + probe  │
  │ local    │  │ local    │  │ local    │
  │ partitions│  │ partitions│  │ partitions│
  └──────────┘  └──────────┘  └──────────┘

  No shuffle of the large side!

Key Takeaway

A Broadcast Join sends the entire small table to every executor, where each builds an in-memory hash table and probes it with the large table's local partitions. The crucial benefit is that the large table never shuffles, which eliminates the most expensive step of a join.

In PySpark, you request it by wrapping the small side: from pyspark.sql.functions import broadcast; big.join(broadcast(small), "key"). Spark also broadcasts automatically when a side is under the 10 MB threshold, so the explicit hint is for when statistics are stale or when you want certainty.

At the National Bank of India, this was the standard way we enriched our billion-row transactions table with the 50-row branch dimension. Every executor got a copy of the branch list and looked up branch names locally with no shuffle of the big table. When a stale statistic once caused Spark to plan a sort-merge join, adding the explicit broadcast hint fixed it immediately.

The caution is that broadcasting is not free: the small side is materialized on the Driver and copied to every executor, and deserialized size can be several times the on-disk size. Only broadcast genuinely small tables.

Remember one sentence: a Broadcast Join copies the small table to every executor so the large table joins locally with no shuffle, and in PySpark you request it with broadcast(small).

---

Share
Use ← → arrow keys to navigate