Broadcast variables

01 · What it is

Two pipes across the driver↔executor boundary

Most values in your code reach the executors by being captured in a closure and serialized ad-hoc with every task. Spark's two shared variables cross that boundary in a controlled way instead — one going out, one coming back:

• A broadcast variable is a read-only value you build on the driver; Spark caches it once per executor, so every task on that executor reads the same in-memory copy. Create it with sc.broadcast(value); read it on workers with bv.value.
• An accumulator is a write-only-on-workers counter: tasks only .add() to it, and only the driver reads the final .value. Create it with sc.accumulator(0); tasks do acc.add(1).

analogy Think of them as opposite-direction pipes. A broadcast variable pushes data out to every executor; an accumulator pulls aggregated numbers back to the driver. The deep dive below covers each pipe once; the diagrams show them running.

02 · Why it matters

What each pipe saves you

• Without a broadcast variable, a lookup dict/set/model referenced in a closure is serialized and shipped with every task — 10,000 tasks on a stage = 10,000 copies of the same 50 MB map crossing the network. With a broadcast variable it ships once per executor and is reused by all its tasks: far less network, GC churn, and duplicated memory.
• Without an accumulator, getting a number out of distributed tasks (bad-record count, rows-skipped) needs a whole extra pass over the data. An accumulator lets tasks report a number as a side-effect of work you're already doing — no extra job.
• Interviewers probe both: "How do you ship a 200 MB model to your UDF without re-sending it per task?" (broadcast variable) and "Can you trust a counter incremented inside a map?" (no — the exactly-once guarantee holds only inside actions).

03 · How it works — deep dive

Ship a value out; count a number back

Each pipe is explained here once — its mechanism, its one hard constraint, and the trap it hides. The diagrams that follow show these ideas run; the tables and gotchas cross-reference them. Neither is re-explained.

3.1 · Broadcast variable — a read-only value, once per executor

analogy Like handing every cashier (executor) their own laminated price-list at the start of the shift, instead of faxing it to every customer (task). One copy per till, reused all day.

• Mechanism. bv = sc.broadcast(value) on the driver serializes the value and ships it to each executor once (a torrent-like fan-out); the executor caches it in Storage memory / the block manager, and every task reads the same bv.value. The docs define it as "a read-only variable cached on each machine rather than shipping a copy of it with tasks."
• Read-only. Workers only read bv.value. Mutating it on a worker changes nothing anywhere else — it's broadcast once, not kept in sync.
• Why it beats a closure variable (Rung 2). A captured closure var is serialized once per task (one copy per partition); a broadcast var is serialized once per executor. On a wide stage that's thousands of copies versus one.
• Lifecycle. bv.unpersist() drops the cached copies (re-broadcast if reused); bv.destroy() removes it permanently (using it afterwards is an error). Release it — it pins Storage memory on every executor.
• The one hard constraint. The value must fit on the driver (where it's built) and in each executor's memory. It's static and read-only; a per-row-changing lookup is the wrong fit.

use case A 30 MB country_code → region dict (or a small ML model) referenced in a Python UDF scoring 5 billion rows — broadcast once per executor rather than shipping it with every task.

3.2 · Accumulator — count on workers, read on the driver

analogy Like a tally counter at a stadium turnstile — each gate (task) clicks it as people pass, but only the control room (driver) reads the total. Gates can add; they can't read the running count.

• Mechanism. acc = sc.accumulator(0) creates a shared counter on the driver — PySpark's only accumulator constructor (there is no sc.longAccumulator(name) in PySpark; that's Scala/Java-only). Tasks call acc.add(n) (or acc += n); each executor keeps a local partial and sends its contribution back, where Spark merges them on the driver. Only the driver reads acc.value.
• Write-only on workers. A task that reads acc.value sees only its own partial — never the global total. That asymmetry is what makes the merge associative and cheap.
• Lazy. The update only happens when an action runs the transformation containing it. No action → the count stays at the initial value.
• The exactly-once caveat (Rung 3 — the whole interview). The docs guarantee exactly-once only for updates inside actions (e.g. foreach): "In transformations, users should be aware that each task's update may be applied more than once if tasks or stages are re-executed." A map/filter that increments an accumulator may run twice (a retried/speculative task, a recomputed stage, or a second action over an uncached DataFrame) → an inflated count.
• Trade-off. Great for cheap side-channel metrics; not a substitute for a real aggregation. For correctness-critical counts, use groupBy().count() / filter().count().

use case Counting malformed records while writing a cleaned dataset — bad_rows.add(1) inside the action that writes (foreachPartition), then log bad_rows.value on the driver, with no second pass.

Figure A · the two pipes side by side

Once-per-executor vs. actions-only, in two panels

Two independent panels, each isolating the one thing that trips people up. Left — a broadcast variable: drag the tasks-per-executor and watch the copy count stay at one per executor, not one per task. Right — an accumulator: run it inside an action or a transformation, then re-run to see the count stay correct or double. (Where the data physically travels across nodes is Figure B.)

04 · The code — and how to verify it

Three moves, each with its proof

Every snippet below is paired with how to prove it worked — the Spark-UI signal or the cross-check. Apply, then verify; never assume.

Broadcast a read-only lookup, read it in a UDF

python# DRIVER: build the small lookup once, then broadcast it (read-only) to every executor.
lookup = {"US": "NA", "CA": "NA", "DE": "EU", "IN": "APAC"}
bv = sc.broadcast(["US", "CA", "DE", "IN"])   # any plain Python object — shipped ONCE per executor
bv = sc.broadcast(lookup)                              # here: a dict; read on workers via bv.value

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# WORKER: each task reads the SAME cached copy via bv.value — no per-task re-send.
@udf(StringType())
def to_region(code):
    return bv.value.get(code, "UNKNOWN")             # bv.value is READ-ONLY on the worker

regions = orders.withColumn("region", to_region("country_code"))

# VERIFY — it's a *variable*, not a plan node, so check the Spark UI, not .explain():
#   Spark UI -> Storage tab shows ONE "Broadcast" block sized ~your value, NOT x num_tasks.
bv.unpersist()   # release when done (re-broadcast if reused) …
# bv.destroy()   # … or remove permanently (using it afterwards raises an error)

Count bad records inside an action (exactly-once)

python# DRIVER: sc.accumulator(value) is PySpark's ONLY accumulator constructor. It is UNNAMED.
# (Scala/Java have sc.longAccumulator("name"); PySpark has no sc.longAccumulator.)
bad_rows = sc.accumulator(0)

def validate_and_write(rows):
    for r in rows:
        if r["amount"] is None:
            bad_rows.add(1)                       # WORKER: only .add() — never read .value here
        # … write/emit the good row …

# Inside an ACTION (foreachPartition) -> each task's update applied EXACTLY ONCE.
orders.rdd.foreachPartition(validate_and_write)

print("bad rows:", bad_rows.value)                # DRIVER ONLY reads the merged total

# VERIFY — PySpark's sc.accumulator() is UNNAMED, so it is typically NOT surfaced by name
#   in the Spark UI Stages -> Accumulators view. Cross-check against a real aggregation:
print("true count:", orders.filter(orders.amount.isNull()).count())   # must match

The anti-pattern: incrementing inside a transformation

python# WRONG: incrementing inside a TRANSFORMATION (lazy + may re-run) — NOT exactly-once.
seen = sc.accumulator(0)
counted = orders.rdd.map(lambda r: (seen.add(1), r)[1])   # transformation, not action
counted.count()        # runs once  -> seen == N …
counted.count()        # runs AGAIN (not cached) -> seen == 2N  (inflated)

# A retried/speculative task or a recomputed stage can double-count even with ONE action.
# VERIFY / FIX — use a true aggregation and compare; they will NOT match after two actions:
print(orders.count())                       # the real count — trust this, not `seen`

Figure B · where it travels

See it run across the cluster

This figure isolates a different thing from Figure A: not the counters, but where data physically moves across 1 driver + 3 executors. Part A ships a broadcast variable out — once per executor, reused by every task on it. Part B runs an accumulator: tasks .add() to a local partial, the deltas flow back to the driver, and only the driver reads .value. Use Prev / Play / Next.

05 · Comparison table

The two pipes, dimension by dimension

Read this by the property you care about — it answers "which pipe, and what's its catch?" at a glance. (The why is §3; this is the which.)

06 · Edge cases & limitations

Where it gets subtle

New nuance an interviewer probes — not a re-run of §3:

• A broadcast value that's almost too big. It must fit on the driver and in each executor's memory; a 2 GB "small" lookup pressures Storage memory (region R, Lesson 04) and can OOM. Above budget, rethink the design — a join or a bucketed lookup table.
• Mutating a broadcast value on a worker. Read-only; local mutations are silently discarded and never propagate. To change the lookup, re-broadcast a new variable.
• Forgetting to release. A long-lived broadcast pins Storage memory on every executor and pressures GC (Lesson 10). Call unpersist()/destroy() when done.
• Accumulator inside a transformation that runs twice. Two actions over the same uncached DataFrame recompute and double the count; even one action can double-count on a retried/speculative task or recomputed stage.
• Reading an accumulator on a worker. Illegal/meaningless — you get a partial, not the total. Read .value on the driver, after the action.

Hard limits worth stating plainly:

• Both are RDD/SparkContext-level APIs. Used via sc.broadcast / sc.accumulator and read inside UDFs/rdd functions. For pure DataFrame work, prefer DataFrame-native tools — the broadcast() join hint and built-in aggregations.
• Spark Connect / serverless. SparkContext.broadcast and SparkContext.accumulator are not supported on Spark Connect. On Databricks serverless / shared-access clusters sc isn't available the same way — use a broadcast join hint and DataFrame aggregations there. They work on classic single-user DBR clusters.
• No exactly-once inside transformations — by design. Don't build correctness-critical counts on accumulators; use a real aggregation.

07 · Common mistakes

Gotchas — the mistake and its one-line fix

08 · References

Sources

• Apache Spark — RDD Programming Guide (Broadcast Variables & Accumulators, exactly-once caveat)
• Apache Spark — SQL Performance Tuning (broadcast join — keep distinct from the variable)
• Apache Spark — Configuration (autoBroadcastJoinThreshold — the join threshold, not the variable)
• Azure Databricks — Adaptive Query Execution (runtime broadcast-join behavior)