Definition
Broadcast variable and broadcast join operate at different layers.
- Broadcast Variable – Created with
sc.broadcast(value), distributes any read-only object to every executor once. Used for map-side lookups in RDD code, UDFs, or distributing non-tabular objects like ML models. - Broadcast Join – A high-level DataFrame/SQL join strategy where the optimizer broadcasts a small table so the large side is not shuffled. Managed automatically by Catalyst and AQE.
- Same machinery – Both use Spark's broadcast mechanism under the hood, but at different abstraction levels.
- Memory caution – Both require the broadcast data to fit in executor memory.
Use broadcast variables for non-relational lookups; use broadcast joins for relational joins.
A broadcast variable is like giving every clerk at the National Bank of India a personal reference card, say a currency-conversion chart, that they can glance at while processing anything. A broadcast join is specifically the act of photocopying one list so clerks can match two lists locally. The reference card is a general tool you hand out and use however you like; the broadcast join is a specific matching optimization the system performs for you.
Understanding the Concepts
What is a broadcast variable?
A broadcast variable solves the problem of shipping large read-only data to executors efficiently. Without it, Spark serializes the object into every task closure, which is wasteful with thousands of tasks.
from pyspark import SparkContext
sc = SparkContext.getOrCreate()
# Currency conversion rates for the National Bank of India
currency_rates = {
"USD": 83.50,
"EUR": 91.20,
"GBP": 106.30,
"JPY": 0.56
}
# Broadcast the rates to all executors (sent once, cached)
rates_broadcast = sc.broadcast(currency_rates)
# Use in a map operation: each task reads from the cached copy
txns_rdd = sc.textFile("/data/bank/forex_transactions.csv")
def convert_to_inr(line):
parts = line.split(",")
currency = parts[2]
amount = float(parts[3])
rate = rates_broadcast.value.get(currency, 1.0)
return (parts[0], parts[1], amount * rate) # (txn_id, account_id, amount_inr)
inr_txns = txns_rdd.map(convert_to_inr)
inr_txns.take(5)
What is a broadcast join?
A broadcast join is a join execution strategy chosen by the Catalyst Optimizer for DataFrame and SQL joins. The small table is broadcast to all executors and the large table is probed locally with a hash table, avoiding a shuffle.
from pyspark.sql.functions import broadcast
txns = spark.read.parquet("/data/bank/transactions") # 500 GB
branches = spark.read.parquet("/data/bank/branches") # 5 MB
# Broadcast join: branches is broadcast, txns is not shuffled
result = txns.join(broadcast(branches), "branch_code")
result.explain()
# BroadcastHashJoin [branch_code]
Broadcast Variable:
- You create it manually with sc.broadcast()
- You read it inside map/filter/UDF with variable.value
- General purpose: any read-only object
Broadcast Join:
- The optimizer creates it automatically (or you hint)
- You do not manage the distribution
- Specialized for join operations
When do you use a broadcast variable?
Use broadcast variables when you need a shared read-only object inside transformations that are not relational joins.
from pyspark.sql.functions import udf
from pyspark.sql.types import DoubleType
import java.math.BigDecimal
# Broadcast a lookup map for use inside a UDF
branch_risk_scores = {
"MUMBAI-01": 0.85,
"DELHI-02": 0.72,
"KOLKATA-03": 0.90
}
risk_broadcast = spark.sparkContext.broadcast(branch_risk_scores)
@udf(DoubleType())
def get_risk_score(branch_code):
return risk_broadcast.value.get(branch_code, 0.5)
txns = spark.read.parquet("/data/bank/transactions")
enriched = txns.withColumn("risk_score", get_risk_score("branch_code"))
enriched.show()
# Broadcast a machine learning model for inference
import pickle
with open("/models/fraud_model.pkl", "rb") as f:
fraud_model = pickle.load(f)
model_broadcast = sc.broadcast(fraud_model)
@udf(DoubleType())
def predict_fraud(amount, branch_code):
model = model_broadcast.value
return float(model.predict([[amount, hash(branch_code)]]))
When do you use a broadcast join?
Use broadcast joins for relational joins of DataFrames or SQL tables where one side is small enough.
# Relational join: broadcast join is the right tool
txns = spark.read.parquet("/data/bank/transactions")
branches = spark.read.parquet("/data/bank/branches") # 5 MB
# Auto-broadcast (optimizer decides based on size)
result = txns.join(branches, "branch_code")
# Explicit broadcast hint
result = txns.join(broadcast(branches), "branch_code")
# SQL equivalent
spark.sql("""
SELECT /*+ BROADCAST(b) */
t.account_id, t.amount, b.branch_name
FROM transactions t
JOIN branches b ON t.branch_code = b.branch_code
""")
How do they compare?
| Feature | Broadcast Variable | Broadcast Join |
|---|---|---|
| Level | Low-level (SparkContext) | High-level (DataFrame/SQL) |
| Created by | Developer manually | Optimizer automatically |
| Data type | Any serializable object | DataFrame/Table |
| Use case | Map-side lookups, UDFs, models | Relational joins |
| AQE integration | No | Yes (dynamic switching) |
| Catalyst optimization | No | Yes (full plan optimization) |
Why prefer broadcast joins over broadcast variables?
In modern DataFrame-centric Spark, most enrichment is better expressed as a DataFrame join that becomes a broadcast join. The optimizer handles sizing, join strategy, and AQE runtime switching.
# Broadcast variable approach (old, manual)
rates_map = {"USD": 83.50, "EUR": 91.20}
rates_bc = sc.broadcast(rates_map)
@udf(DoubleType())
def convert(currency):
return rates_bc.value.get(currency, 1.0)
result = txns.withColumn("rate", convert("currency"))
# Broadcast join approach (modern, optimized)
rates_df = spark.createDataFrame(
[("USD", 83.50), ("EUR", 91.20)],
["currency", "rate"]
)
result = txns.join(broadcast(rates_df), "currency")
# Optimizer handles everything, AQE can adjust
When are broadcast variables still valuable?
Broadcast variables remain essential for RDD-level code, distributing non-tabular objects (ML models, rule sets, configuration), and inside UDFs where a relational join is not natural.
# Still valuable: distributing a non-tabular object
rule_engine = load_fraud_rules() # Complex object, not a DataFrame
rule_broadcast = sc.broadcast(rule_engine)
@udf(StringType())
def apply_rules(amount, branch_code, txn_type):
engine = rule_broadcast.value
return engine.evaluate(amount, branch_code, txn_type)
result = txns.withColumn("risk_level", apply_rules("amount", "branch_code", "txn_type"))
What should a senior engineer know?
- Both share the same memory caution: the broadcast data must fit in executor memory.
- A broadcast variable is read-only by design: mutating it would break the shared-copy guarantee.
- In modern Spark, reach for broadcast joins first. Only use broadcast variables for non-tabular objects or RDD-level code.
- AQE can dynamically switch a sort-merge join to a broadcast join at runtime, but it cannot help broadcast variables: those are manual.
# Decision tree:
# Is the operation a relational join?
# YES → use broadcast join (DataFrame/SQL)
# NO → is it a lookup inside a UDF or RDD map?
# YES → use broadcast variable
# NO → use a regular DataFrame operation
Visual Overview
Broadcast Variable vs Broadcast Join:
Broadcast Variable (low-level):
┌──────────────┐
│ sc.broadcast( │
│ rates_map │──► Sent to every executor once
│ ) │ Cached in executor memory
└──────────────┘
│
▼
┌──────────────────────────────┐
│ RDD.map(lambda: use .value) │
│ UDF(lambda: use .value) │
│ General-purpose lookup │
└──────────────────────────────┘
Broadcast Join (high-level):
┌──────────────┐
│ Small Table │──► Optimizer broadcasts automatically
│ (DataFrame) │ Built into hash table on executors
└──────────────┘
│
▼
┌──────────────────────────────┐
│ Large Table probed locally │
│ No shuffle of large side │
│ Managed by Catalyst + AQE │
└──────────────────────────────┘
Key Takeaway
A broadcast variable and a broadcast join both distribute data to every executor, but they live at different layers. A broadcast variable is a low-level primitive you create with sc.broadcast(), wrapping any read-only object so Spark ships it once and caches it. You use it for map-side logic: a currency chart, a configuration object, or a machine learning model inside a UDF. A broadcast join is a high-level query optimization where the Catalyst Optimizer broadcasts a small table for a join, and AQE can switch to it at runtime.
In modern DataFrame-centric Spark, most enrichment is better written as a join that becomes a broadcast join, because the optimizer manages sizing, strategy, and runtime switching. Broadcast variables remain essential for RDD-level code and for distributing non-tabular objects that do not fit the relational model.
At the National Bank of India, relational lookups like branch names and currency rates were expressed as broadcast joins. The fraud detection model, a complex serialized object, was distributed as a broadcast variable for use inside a UDF. Both carry the same memory caution: the broadcast data must fit comfortably in executor memory.
The decision is simple: if the operation is a relational join and one side is small, use a broadcast join. If you need a shared read-only object inside a transformation that is not a join, use a broadcast variable.
Remember one sentence: broadcast variables distribute general-purpose read-only objects for manual lookups, while broadcast joins are the optimizer-managed join strategy: prefer broadcast joins for relational work and broadcast variables for non-tabular objects.
---