Broadcast join vs shuffle join

Description

Module 4 showed both behaviors, one at each end, without connecting them to an explicit criterion: orders_df.join(dim_store, "store_id") over forty rows produced BroadcastHashJoin, with no shuffle Exchange at all; a self-join of orders_at_scale_df over ten million rows produced SortMergeJoin, with Exchange on both sides. This lesson names the exact criterion that decides between the two — a single setting, with a default value measured in bytes — and verifies it by quoting Spark's official documentation, word for word.

Connection to the module. This lesson is the conceptual foundation for lessons 3 and 4: without this lesson's explicit criterion, "why did Spark pick BroadcastHashJoin here" would stay a black box. With it, lessons 3 and 4 only have to confirm, with .explain(), that the criterion holds exactly as predicted.

An analogy: photocopying the directory, or stopping every truck

Pick back up lesson 1's analogy. A delivery company needs every truck to know which address corresponds to each customer code. If the complete customer directory fits in a few pages, the decision is obvious: photocopy it whole, and hand a copy to every truck — each driver resolves the lookup with what they're already carrying, without asking anything of the other trucks or stopping the route. That's a broadcast join: Spark copies the small table whole — serialized, compressed in memory — to every partition on the large side, and each partition resolves the match locally.

But a reasonable logistics manager wouldn't photocopy a two-million-customer directory just so each of two hundred trucks can carry its own complete copy — the copies' total weight would far exceed the original directory's weight. In that case, the correct decision is the other one: stop every truck, reorganize the orders on both groups (the large one and the one that's no longer so small) by the same criterion, and match them once reorganized. That's a sort-merge join — the shuffle you already measured in module 4, now applied specifically to resolving a JOIN. This lesson's criterion answers, precisely, the question: how big is "too big to photocopy"?

Worked example: the criterion, in Spark's own configuration

Step 1 — The setting that decides everything

# broadcast_threshold.py
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").getOrCreate()

threshold = spark.conf.get("spark.sql.autoBroadcastJoinThreshold")
print(f"spark.sql.autoBroadcastJoinThreshold = {threshold}")

threshold_bytes = int(threshold.rstrip("b"))
threshold_mb = threshold_bytes / (1024 * 1024)
print(f"In bytes: {threshold_bytes:,}")
print(f"In MB: {threshold_mb:.1f}")

spark.stop()

What to expect. Running python3 broadcast_threshold.py, the output is exactly this (executed in this run, PySpark 4.2.0):

spark.sql.autoBroadcastJoinThreshold = 10485760b
In bytes: 10,485,760
In MB: 10.0

This is the full criterion, in a single figure: 10 MB. Spark's official documentation (SQL Performance Tuning) describes it like this, word for word:

"Configures the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join. By setting this value to -1, broadcasting can be disabled."

Notice two things in the official text. First, "maximum size in bytes for a table" — the criterion measures the complete table's estimated size in bytes, not its row count, nor a single column's size. Second, "By setting this value to -1, broadcasting can be disabled" — Spark explicitly documents -1 as the official way to disable broadcast join entirely, exactly the mechanism lesson 4 is going to use to force a SortMergeJoin.

Step 2 — Why dim_store and dim_product fall, with no doubt at all, under that threshold

# dim_sizes.py
import os

for fname in ["dim_store.csv", "dim_product.csv"]:
    size_bytes = os.path.getsize(fname)
    print(f"{fname}: {size_bytes} bytes")

threshold_bytes = 10 * 1024 * 1024
print(f"\nDefault threshold: {threshold_bytes:,} bytes (10 MB)")

What to expect (executed in this run):

dim_store.csv: 79 bytes
dim_product.csv: 138 bytes

Default threshold: 10,485,760 bytes (10 MB)

dim_store.csv (three stores) and dim_product.csv (four products) together weigh less than two hundred bytes — a tiny fraction of the ten-megabyte threshold. There's no ambiguity at all in the decision Spark is going to make: both tables fit, with plenty of room to spare, to be copied whole to every partition of fact_orders_at_scale, regardless of that large side having ten million rows. Lesson 3 confirms this with .explain(), reading the real physical plan.

Diagram: the criterion, in a single decision

flowchart TD
    A["JOIN between two tables"] --> B{"Is the smaller table's\nestimated size in bytes below\nautoBroadcastJoinThreshold?\n(10 MB by default)"}
    B -->|"Yes -- dim_store (79 B),\ndim_product (138 B)"| C["BroadcastHashJoin\nBroadcastExchange copies the whole\nsmall table to every partition.\nNO shuffle Exchange."]
    B -->|"No -- no table\nfits for broadcast"| D["SortMergeJoin\nExchange hashpartitioning\non BOTH sides of the JOIN."]

    style C fill:#9c6,stroke:#333
    style D fill:#f96,stroke:#333

Going deeper: what the criterion does NOT say, and why it matters

The spark.sql.autoBroadcastJoinThreshold threshold decides based on estimated size, not on the real size measured with perfect precision — and that word, "estimated," carries a practical consequence worth knowing before blindly trusting the default behavior. When Spark reads a CSV or Parquet file, it keeps statistics — size on disk, approximate row count — that it uses to estimate how much a table will weigh in memory once deserialized. For dim_store.csv and dim_product.csv, the difference between "size on disk" and "real size in memory" is irrelevant — both figures are so far below the threshold that no estimation error would change the decision. But in an edge case — a table that weighs, say, 9.8 MB on disk but expands to 15 MB in memory because of how Spark represents its data types — the estimate could be wrong, and Spark could attempt a broadcast that in practice turns out more expensive than expected. That's why Spark also documents explicit JOIN hints (BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL) for cases where you want to force a strategy without depending on automatic estimation — a hint tells Spark "use this strategy with this table, no matter what the statistics say." This guide doesn't use JOIN hints (aside from forcing the full threshold in lesson 4), but it's worth knowing they exist for the day you work with tables whose real size sits close to the limit.

It's also worth noting something you already saw in module 4, now under its correct name: the official documentation mentions that Adaptive Query Execution can convert a planned SortMergeJoin into a BroadcastHashJoin at runtime, if the real statistics (measured after reading the data, not estimated beforehand) show one side ended up small enough — for example, after an aggressive .filter() sharply shrank a table that originally looked large. This is an additional optimization on top of this lesson's static criterion, and module 6 revisits it in more depth when covering AQE.

Common mistakes

Thinking the threshold is measured in row count, not bytes. What happens: someone, hearing "small table" to describe when Spark broadcasts, assumes the criterion is something like "fewer than a thousand rows," and is surprised when a table with only a hundred rows does not get broadcast. Why it happens: in everyday language, "small table" tends to evoke "few rows," and it's easy not to distinguish that from "few bytes." How to spot it: if you can't explain why a hundred-row table with a very long text column could not qualify for broadcast, while a million-row table with tiny columns could qualify, you're missing the criterion's exact definition. How to fix it: remember this lesson's official quote, word for word: "maximum size in bytes" — the criterion is always about estimated bytes, never about row count. dim_store and dim_product qualify for broadcast both by low row count and by low byte count, so for Kiosko this distinction doesn't change the conclusion — but on real data, it can.

Confusing "disabling broadcast" with "always making the join slower." What happens: someone, after seeing that -1 disables broadcast, assumes forcing SortMergeJoin is, by definition, a bad performance decision in any scenario. Why it happens: module 4 already established that an Exchange has a real cost, and it's easy to generalize "shuffle = bad" with no nuance. How to spot it: if your reasoning doesn't distinguish between "when a broadcast join is possible" (a question of whether the table fits) and "when a broadcast join is still the best decision" (a broader question, depending on the whole cluster), you're missing the nuance. How to fix it: lesson 4 forces SortMergeJoin for a purely pedagogical purpose — comparing both plans side by side, over the same join — not because it's the recommended decision for dim_store/dim_product in production. With tables as small as Kiosko's, letting Spark use the default broadcast is, almost always, the right call.

Assuming this lesson's criterion applies equally to any kind of JOIN. What happens: someone generalizes the spark.sql.autoBroadcastJoinThreshold criterion to any operation that combines two DataFrames, including a union() or a .crossJoin(), without checking that the documentation applies specifically to equality-condition joins (equi-joins). Why it happens: "combining two tables" sounds like a single category, and it's easy not to distinguish among the different operations Spark offers for it. How to spot it: if you can't name, specifically, which type of JOIN this lesson's official quote is describing, check the text's full context — the SQL Performance Tuning section introducing it explicitly discusses equality-key JOIN strategies. How to fix it: every join in this guide — the ones inherited from data-modeling-for-analytics-guide, and the ones in this module — is an equi-join with the abbreviated form (.join(df, "column")), exactly the case this criterion describes. Don't assume the same threshold applies, without checking, to operations that aren't an equi-join.

Exercises

Exercise 1 — Calculate how many times dim_product.csv would fit within the default threshold. Using dim_product.csv's size in bytes from the worked example (138 bytes), calculate how many copies of that file would fit, in total, within the default threshold's 10,485,760 bytes.

dim_product_bytes = 138
threshold_bytes = 10 * 1024 * 1024

times = threshold_bytes // dim_product_bytes
print(f"dim_product.csv ({dim_product_bytes} bytes) would fit {times:,} times within the threshold")

dim_product.csv (138 bytes) would fit 75,983 times within the threshold

Exercise 2 — Predict and verify: what happens if you try spark.conf.get() for a setting that doesn't exist? Without running anything, predict what would happen if you called spark.conf.get("spark.sql.this.setting.does.not.exist"), with no default value. Verify your prediction.

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").getOrCreate()

try:
    spark.conf.get("spark.sql.this.setting.does.not.exist")
except Exception as e:
    print(type(e).__name__, "-", str(e)[:150])

# with a default value, it doesn't fail
safe_value = spark.conf.get("spark.sql.this.setting.does.not.exist", "not configured")
print(f"With default value: {safe_value}")
spark.stop()

SparkNoSuchElementException - [SQL_CONF_NOT_FOUND] The SQL config "spark.sql.this.setting.does.not.exist" cannot be found. Please verify that the config exists. SQLSTATE: 42K0I
With default value: not configured

Exercise 3 — Explain, without code, why -1 (and not 0) is the value that disables broadcast join. In 2-3 sentences, explain why it makes sense, from a configuration-design standpoint, that Spark uses -1 (a value that could never be a real size in bytes) instead of 0 to represent "never broadcast."

Summary and next step

This lesson named, with the exact official quote, the complete criterion that decides between BroadcastHashJoin and SortMergeJoin: spark.sql.autoBroadcastJoinThreshold, 10 MB (10,485,760 bytes) by default, measured on the estimated size in bytes of the JOIN's smaller table — never on row count. You confirmed that dim_store.csv (79 bytes) and dim_product.csv (138 bytes) sit, by several orders of magnitude, below that threshold, and that -1 is the officially documented value for disabling broadcast entirely.

Before moving on you should be able to: cite the threshold's default value from memory (10 MB); explain why the criterion measures bytes, not rows; and explain what -1 does as this setting's value, and why that specific value makes sense.

Lesson 3 confirms, with .explain() run over the complete fact_orders_at_scale (ten million rows) against dim_store and dim_product, that Spark makes exactly the decision this criterion predicts: BroadcastHashJoin, with no shuffle Exchange at all.

Resources

• Apache Spark — SQL Performance Tuning, "Broadcast Hash Join" section and the spark.sql.autoBroadcastJoinThreshold configuration properties table (this lesson's exact quote: "Configures the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join. By setting this value to -1, broadcasting can be disabled"). spark.apache.org/docs/latest/sql-performance-tuning.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this lesson's exact specification and its place in module 5.
• data-modeling-for-analytics-guide's DESIGN doc — the source of dim_store and dim_product, the two tables whose size this lesson measured in bytes. src/guides/data-modeling-for-analytics-guide/DISENO.md