Reading a broadcast join in `.explain()`

Description

Lesson 2 established the criterion: dim_store (79 bytes) and dim_product (138 bytes) sit, without a doubt, below the default 10 MB threshold. This lesson confirms with real evidence that Spark makes exactly the decision that criterion predicts — not over Kiosko's forty real rows, which you already saw in module 4, but over the complete fact_orders_at_scale: ten million synthetic rows, joined against the same two tiny dimension tables. The physical plan is going to show BroadcastHashJoin, with no shuffle Exchange at all, regardless of the large side having ten million rows.

Connection to the module. This lesson introduces no new concept — it applies lesson 2's criterion to this guide's at-scale dataset, and reads the evidence in .explain(). Lesson 4 is going to repeat this exact same JOIN, with the threshold disabled, so you can compare both plans side by side.

An analogy: the directory, photocopied once more, now for a huge warehouse

In lesson 2, the analogy compared photocopying a small directory against reorganizing the whole cargo. This lesson applies that same decision at a much larger scale on one side only: instead of seven trucks with forty boxes, now it's twelve trucks carrying ten million boxes. The question the logistics manager has to answer stays the same — is the directory small enough to photocopy? — and the answer doesn't change based on how many boxes the large side carries: if the directory (dim_store, dim_product) still weighs the same, it still qualifies for photocopying, regardless of whether there are forty boxes or ten million waiting on the other side.

Worked example: M4 lesson 2's same join, now at full scale

Step 1 — Read fact_orders_at_scale, dim_store, and dim_product

# broadcast_at_scale.py
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)

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

scale_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("franchise_id", IntegerType(), False),
    StructField("store_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),
    StructField("order_ts", TimestampType(), False),
])
dim_store_schema = StructType([
    StructField("store_id", StringType(), False),
    StructField("store_name", StringType(), False),
    StructField("city", StringType(), False),
])
dim_product_schema = StructType([
    StructField("product_id", StringType(), False),
    StructField("product_name", StringType(), False),
    StructField("category", StringType(), False),
    StructField("unit_cost", DoubleType(), False),
])

orders_at_scale_df = spark.read.csv(
    "kiosko_orders_at_scale.csv", schema=scale_schema, header=True, enforceSchema=False,
)
dim_store_df = spark.read.csv("dim_store.csv", schema=dim_store_schema, header=True, enforceSchema=False)
dim_product_df = spark.read.csv("dim_product.csv", schema=dim_product_schema, header=True, enforceSchema=False)

fact_orders_at_scale_df = orders_at_scale_df.withColumn(
    "revenue", F.col("quantity") * F.col("unit_price")
)

Step 2 — The join, and its physical plan

joined_store = fact_orders_at_scale_df.join(dim_store_df, "store_id")

print("=== fact_orders_at_scale_df.join(dim_store_df, 'store_id') -- physical plan ===")
joined_store.explain()

What to expect. Running python3 broadcast_at_scale.py, the output is exactly this (executed in this run, over the complete fact_orders_at_scale, 10,000,000 rows):

=== fact_orders_at_scale_df.join(dim_store_df, 'store_id') -- physical plan ===
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [store_id#2, order_id#0, franchise_id#1, product_id#3, quantity#4, unit_price#5, order_ts#6, revenue#15, store_name#8, city#9]
   +- BroadcastHashJoin [store_id#2], [store_id#7], Inner, BuildRight, false, false
      :- Project [order_id#0, franchise_id#1, store_id#2, product_id#3, quantity#4, unit_price#5, order_ts#6, (cast(quantity#4 as double) * unit_price#5) AS revenue#15]
      :  +- Filter isnotnull(store_id#2)
      :     +- FileScan csv [order_id#0,franchise_id#1,store_id#2,product_id#3,quantity#4,unit_price#5,order_ts#6] Batched: false, DataFilters: [isnotnull(store_id#2)], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/.../kiosko_orders_at_scale.csv], PartitionFilters: [], PushedFilters: [IsNotNull(store_id)], ReadSchema: struct<order_id:string,franchise_id:int,store_id:string,product_id:string,quantity:int,unit_price...
      +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=26]
         +- Filter isnotnull(store_id#7)
            +- FileScan csv [store_id#7,store_name#8,city#9] Batched: false, DataFilters: [isnotnull(store_id#7)], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/.../dim_store.csv], PartitionFilters: [], PushedFilters: [IsNotNull(store_id)], ReadSchema: struct<store_id:string,store_name:string,city:string>

Not a single Exchange hashpartitioning anywhere in the plan — zero reorganization on the ten-million-row side. Instead, the strategy is exactly BroadcastHashJoin, with BuildRight (the "small" table, the right side of the .join() call, is the one that gets copied), and BroadcastExchange building that copy once to distribute it to every partition. This confirms, with real evidence over ten million rows, the exact prediction you made in lesson 2 just by looking at dim_store.csv's byte count.

Step 3 — Chaining the second join, against dim_product

joined_full = joined_store.join(dim_product_df, "product_id")
print("\n=== ...+.join(dim_product_df, 'product_id') -- full physical plan ===")
joined_full.explain()

count_before = fact_orders_at_scale_df.count()
count_after = joined_full.count()
print(f"\ncount before the joins = {count_before}")
print(f"count after both joins = {count_after}")
assert count_before == count_after == 10_000_000
print("Verification: neither BroadcastHashJoin lost a single row -> OK")

spark.stop()

What to expect (executed in this run):

=== ...+.join(dim_product_df, 'product_id') -- full physical plan ===
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [product_id#3, store_id#2, order_id#0, franchise_id#1, quantity#4, unit_price#5, order_ts#6, revenue#15, store_name#8, city#9, product_name#11, category#12, unit_cost#13]
   +- BroadcastHashJoin [product_id#3], [product_id#10], Inner, BuildRight, false, false
      :- Project [store_id#2, order_id#0, franchise_id#1, product_id#3, quantity#4, unit_price#5, order_ts#6, revenue#15, store_name#8, city#9]
      :  +- BroadcastHashJoin [store_id#2], [store_id#7], Inner, BuildRight, false, false
      :     :- Project [order_id#0, franchise_id#1, store_id#2, product_id#3, quantity#4, unit_price#5, order_ts#6, (cast(quantity#4 as double) * unit_price#5) AS revenue#15]
      :     :  +- Filter (isnotnull(store_id#2) AND isnotnull(product_id#3))
      :     :     +- FileScan csv [order_id#0,franchise_id#1,store_id#2,product_id#3,quantity#4,unit_price#5,order_ts#6] Batched: false, DataFilters: [isnotnull(store_id#2), isnotnull(product_id#3)], Format: CSV, ...
      :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=73]
      :        +- Filter isnotnull(store_id#7)
      :           +- FileScan csv [store_id#7,store_name#8,city#9] Batched: false, DataFilters: [isnotnull(store_id#7)], Format: CSV, ...
      +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=77]
         +- Filter isnotnull(product_id#10)
            +- FileScan csv [product_id#10,product_name#11,category#12,unit_cost#13] Batched: false, DataFilters: [isnotnull(product_id#10)], Format: CSV, ...

count before the joins = 10000000
count after both joins = 10000000
Verification: neither BroadcastHashJoin lost a single row -> OK

Two chained BroadcastHashJoins, two BroadcastExchanges — one per dimension table — and zero Exchange hashpartitioning nodes in the entire plan. The ten-million-row side never gets reorganized: every one of its partitions already has, copied in ahead of time, all three of dim_store's rows and all four of dim_product's, so it can resolve both matches on its own. And the row count — ten million before, ten million after two JOINs — confirms no store_id or product_id from fact_orders_at_scale fell out of the match, exactly the same verification discipline you already used in module 3.

Diagram: two BroadcastExchanges, zero shuffle on the large side

flowchart TD
    subgraph Grande["fact_orders_at_scale -- 10,000,000 rows, 12 partitions"]
        P0["Partition 0"]
        P1["Partition 1"]
        P2["... 10 more partitions"]
    end

    subgraph Chicas["Dimension tables, copied whole"]
        S["dim_store\n3 rows, 79 bytes"]
        PR["dim_product\n4 rows, 138 bytes"]
    end

    S -->|"BroadcastExchange"| P0
    S -->|"BroadcastExchange"| P1
    S -->|"BroadcastExchange"| P2
    PR -->|"BroadcastExchange"| P0
    PR -->|"BroadcastExchange"| P1
    PR -->|"BroadcastExchange"| P2

    P0 --> R["joined_full\n10,000,000 rows\n(each partition resolved on its own,\nwith no coordination among the others)"]
    P1 --> R
    P2 --> R

    style S fill:#9c6,stroke:#333
    style PR fill:#9c6,stroke:#333

Going deeper: BuildRight, and why the order of .join() matters for this decision

Notice a detail in the physical plan you hadn't seen yet: BroadcastHashJoin [store_id#2], [store_id#7], Inner, BuildRight, false, false. The word BuildRight tells Spark which of the two sides builds the data structure that's going to be copied (technically, an in-memory hash table, optimized for fast lookups). In fact_orders_at_scale_df.join(dim_store_df, "store_id"), dim_store_df is the argument — the "right" side of the call — and it's also the small side; Spark chose to build that hash table from dim_store_df (BuildRight), exactly the side that makes sense, because it's the one that's going to travel, copied, to every partition on the large side.

This isn't a coincidence of how you wrote the code — it's a decision made by the Catalyst optimizer, which evaluates the estimated size of both sides of the JOIN, regardless of the order you wrote them in. If you had written dim_store_df.join(fact_orders_at_scale_df, "store_id") — flipping the order — Catalyst would still identify dim_store_df as the small side, and would choose BuildLeft instead of BuildRight, but the result — a BroadcastHashJoin with no shuffle Exchange — would be identical. Lesson 2's criterion (size in bytes) decides which strategy to use; BuildLeft/BuildRight only describes which side ends up being the copy, an implementation detail that depends on where the small side physically sits in the expression, not on a business decision.

Common mistakes

Searching for Exchange in the plan and concluding "no shuffle" without distinguishing BroadcastExchange from Exchange hashpartitioning. What happens: someone searches this lesson's plan for the word Exchange, finds it twice (BroadcastExchange), and wrongly concludes there was a hash-reorganization shuffle, just like in module 4. Why it happens: both nodes share the word Exchange in their name, and at first glance they look like the same category of operation. How to spot it: if your reading of the plan doesn't distinguish BroadcastExchange from Exchange hashpartitioning(..., 200), check the node's full name, not just whether it contains the word "Exchange." How to fix it: Exchange hashpartitioning(...) (the one you saw in module 4) reorganizes both sides of an operation by a key, moving real data between partitions — the expensive shuffle. BroadcastExchange builds the small table's copy once and distributes it — a cost, but of a completely different nature, one that doesn't grow with the large side's size. This lesson has BroadcastExchange, not Exchange hashpartitioning — zero reorganization shuffle.

Assuming BroadcastHashJoin means Spark read less data from the large side. What happens: someone, seeing that the ten-million-row side "didn't get reorganized," concludes Spark also didn't have to read it in full. Why it happens: "avoiding the shuffle" sounds like "doing less work overall," and it's easy to extend that idea to the file read itself. How to spot it: if you believe BroadcastHashJoin implies Spark read fewer than all ten million rows of fact_orders_at_scale, check the FileScan in this lesson's plan — it still reads the complete file, with no filter reducing the volume. How to fix it: BroadcastHashJoin avoids reorganizing (moving between partitions) the large side, but every partition on that large side still has to be read and processed in full — the savings are specifically in not having to coordinate those partitions with each other to resolve the JOIN, not in reducing how much gets read.

Expecting the plan to show the exact byte count copied in the BroadcastExchange. What happens: someone looks, in .explain() (the default mode, with no arguments), for an exact figure of how many bytes BroadcastExchange copied, and can't find it anywhere in the plan. Why it happens: module 4 (lesson 5) did show exact figures for bytes moved in a shuffle, but those figures came from the Spark UI (its REST API), not from .explain() — it's easy to expect the same level of detail from the same source. How to spot it: if you're looking for a specific byte count inside plain .explain()'s output, check that you're looking at the right source of evidence. How to fix it: .explain() confirms the plan's structure — which strategy got chosen, which nodes exist — not real execution metrics; for those exact figures (bytes moved, rows processed per stage), the source is the Spark UI or its REST API, exactly as you already saw in module 4, lesson 5.

Exercises

Exercise 1 — Confirm the same BroadcastHashJoin shows up if you join against dim_product first, instead of dim_store. Swap the order of the worked example's two .join() calls — dim_product_df first, dim_store_df second — and confirm with .explain() that the plan still shows BroadcastHashJoin in both cases, with no shuffle Exchange at all.

joined_reordered = fact_orders_at_scale_df.join(dim_product_df, "product_id").join(dim_store_df, "store_id")
joined_reordered.explain()
count_reordered = joined_reordered.count()
print(f"count with reversed order = {count_reordered}")
assert count_reordered == 10_000_000

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [...]
   +- BroadcastHashJoin [store_id#2], [store_id#N], Inner, BuildRight, false, false
      :- Project [...]
      :  +- BroadcastHashJoin [product_id#3], [product_id#M], Inner, BuildRight, false, false
      :     :- Filter [...]
      :     :  +- FileScan csv [...] -- fact_orders_at_scale, 10M rows
      :     +- BroadcastExchange [...] -- dim_product
      +- BroadcastExchange [...] -- dim_store

count with reversed order = 10000000

Exercise 2 — Verify dim_store's store_name and city reached the ten million rows correctly. Using .groupBy("store_id", "store_name", "city").count() over joined_full, confirm every store has exactly the correct name and city, with the expected total count per store.

per_store = joined_full.groupBy("store_id", "store_name", "city").count().orderBy("store_id")
per_store.show(truncate=False)

counts = {r["store_id"]: r["count"] for r in per_store.collect()}
assert counts == {"S01": 4_000_000, "S02": 3_250_000, "S03": 2_750_000}
print("Verification: every store arrived with the correct name/city and the exact count -> OK")

+--------+-------------+--------+-------+
|store_id|store_name   |city    |count  |
+--------+-------------+--------+-------+
|S01     |Kiosko Centro|Bogota  |4000000|
|S02     |Kiosko Norte |Lima    |3250000|
|S03     |Kiosko Sur   |Santiago|2750000|
+--------+-------------+--------+-------+

Verification: every store arrived with the correct name/city and the exact count -> OK

Exercise 3 — Explain, without code, what would happen to the plan if dim_store.csv hypothetically weighed 50 MB instead of 79 bytes. In 2-3 sentences, predict which strategy Spark would choose, and which node would show up in the plan instead of BroadcastExchange.

Summary and next step

This lesson confirmed, with .explain() run over the complete fact_orders_at_scale, that Spark chooses BroadcastHashJoin for the match against dim_store and dim_product, exactly as lesson 2's byte criterion predicted — with no Exchange hashpartitioning node anywhere in the plan, regardless of the large side having ten million rows. You also verified the two chained JOINs preserve the full row count, and saw what BuildRight/BuildLeft means in the physical plan.

Before moving on you should be able to: distinguish BroadcastExchange from Exchange hashpartitioning just by looking at the node's name; explain why BuildRight doesn't depend on the order you wrote the .join() in, but on each table's real size; and predict what would change in the plan if a dimension table grew past the threshold.

Lesson 4 repeats this exact same JOIN, with spark.sql.autoBroadcastJoinThreshold deliberately disabled, so you can compare the SortMergeJoin plan — with its real Exchange — side by side with this lesson's BroadcastHashJoin.

Resources

• Apache Spark — SQL Performance Tuning, "Broadcast Hash Join" section (the reference for the strategy confirmed in this lesson). spark.apache.org/docs/latest/sql-performance-tuning.html.
• Apache Spark — DataFrame.join (the API reference, already cited in module 3, the foundation for this lesson's .join()). spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.join.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this lesson's exact specification: .explain() over fact_orders_at_scale showing BroadcastHashJoin.