Data systems engineering, part 4: Spark joins, broadcast lookup, and hash shuffle

A join has two separate meanings:

Logical meaning:
  Which records should match, and how many output records should exist?

Physical meaning:
  How does the engine arrange records so matching keys can meet?

The first meaning protects correctness. The second determines where the work and cost appear.

This article uses the clean ecommerce orders and customers data from the earlier POCs. The customer table is intentionally small, which lets Spark choose a broadcast join. We then disable that optimization to expose a shuffle join. The business result remains the same while the physical plan changes substantially.

Begin with cardinality, not with Spark settings

The source contract is:

orders:
  one row per order
  key: order_id
  relationship key: customer_id

customers:
  one row per customer
  key: customer_id

The clean fixture contains:

12 orders
6 customers

Every order has one matching customer, and every customer ID appears once in the customer input.

The baseline operation is:

orders.join(
    customers,
    on="customer_id",
    how="inner",
)

An inner join emits one row for every matching pair. Because the customer key is unique, each order matches one customer:

12 orders × 1 customer match per order = 12 output rows

The output grain remains one row per order. Customer attributes such as customer_name, segment, and signup_date are added to each matching order.

That is why the same customer name may appear several times. C001 appears on three orders, so Ada Lovelace appears on those three order rows. The join is not supposed to collapse them into one customer row.

Why duplicate keys are dangerous

A normal join does not deduplicate either side. If the customer table accidentally contained two rows for C001, then the three C001 orders would match both customer rows:

O1001 × 2 customer rows = 2 outputs
O1004 × 2 customer rows = 2 outputs
O1011 × 2 customer rows = 2 outputs

That produces six rows for those three orders. This is join fan-out, or join multiplication.

The example is an analytical consequence of the relational operation, not a claim that the clean fixture contained duplicate customer keys. A production pipeline should test dimension-key uniqueness before using the dimension to calculate revenue or other additive metrics.

The opposite issue is a missing key. An inner join drops an order with no matching customer. A left join preserves the order and supplies null customer attributes. The correct choice depends on the contract: enrichment and reporting may use different trust boundaries.

The smallest runnable join

The learning repository includes a focused script:

cd <learning-repository>/02-spark-batch
bash scripts/run_poc2.sh --join

The join selects order and customer fields without changing the order amount:

return (
    orders.join(customers, on="customer_id", how="inner")
    .select(
        "order_id",
        "customer_id",
        "product_id",
        "order_date",
        "quantity",
        "amount",
        "status",
        "customer_name",
        "segment",
        "signup_date",
    )
)

The observed result was:

orders_read=12
customers_read=6
joined_count=12
distinct_order_count=12

Representative output included:

O1001 | C001 | Ada Lovelace   | 1200.00 | completed
O1004 | C001 | Ada Lovelace   |  400.00 | completed
O1010 | C003 | Linus Torvalds |   75.00 | cancelled

The cancelled order remains in the joined data. The join adds customer context; it does not apply the separate business rule that only completed orders count in the daily-revenue metric.

What Spark chose by default: broadcast lookup

The normal plan reported:

BroadcastHashJoin Inner BuildRight
├── orders scan
└── BroadcastExchange
    └── customers scan

The project reported this automatic broadcast threshold:

auto_broadcast_join_threshold=10485760b

The customer input is far below that threshold, so Spark selected the right-hand customers DataFrame as the build side.

What broadcast means in plain language

Imagine two processing locations. One has order rows and the other has the small customer directory. Spark can make the join easy by copying the small directory to the processing locations that are handling orders:

Customer directory:
  C001 → Ada Lovelace
  C002 → Grace Hopper
  ...

Each order processor:
  Read one order
  Look up customer_id locally
  Emit the enriched order

BroadcastExchange is the physical step that prepares and distributes that small lookup relation. BroadcastHashJoin then uses a hash-based lookup instead of moving the large side of the join by key.

In this local experiment, local[2] means one local Spark process with two local execution threads rather than two remote machines. The logical strategy is still representative: copy the small side instead of shuffling the large side.

The observed normal plan had no Exchange under the orders scan. That is the important physical consequence: Spark did not need to redistribute the orders just to find their customers.

Deliberately disable broadcast

To expose the alternative, the script supports:

bash scripts/run_poc2.sh \
  --join \
  --disable-broadcast

The option sets:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)

This is a teaching control. It does not mean that broadcast joins are bad. It lets us compare two physically different implementations of the same logical join.

The forced-shuffle plan was:

SortMergeJoin Inner
├── Sort
│   └── Exchange
│       └── orders scan
└── Sort
    └── Exchange
        └── customers scan

Both sides now contain:

Exchange
Arguments: hashpartitioning(customer_id, 200)

What hash partitioning is doing

For every row, Spark applies a hash function to the join key:

hash(customer_id) → one of 200 shuffle buckets

All records with the same key follow the same routing rule. If C001 maps to a particular bucket, then these records are sent to that bucket:

O1001 | C001
O1004 | C001
O1011 | C001
C001  | Ada Lovelace

The rows meet because the join key produced the same bucket assignment on both sides.

The number 200 comes from Spark's default shuffle-partition setting. It is the number of logical shuffle buckets, not the number of customer IDs, Parquet folders, files, or simultaneously running workers.

Our dataset has only six customer IDs and eighteen total input records across the two tables. Most of 200 logical buckets therefore have little or no data. This is useful evidence of a configuration mismatch: a default intended for broader workloads can be excessive for a tiny local fixture.

local[2] also matters here. Two local execution slots can work concurrently, but the plan can still describe many logical shuffle buckets. Logical parallelism and concurrent capacity are separate decisions.

Why both sides exchange data

Before the shuffle, Spark cannot assume that either CSV input is already arranged by customer_id in a compatible partitioning scheme.

Conceptually, the data could be spread like this:

Orders:
  O1001 | C001 → processing partition 3

Customers:
  C001  | Ada Lovelace → processing partition 5

The matching rows are separated. Shuffling only one side would not be safe unless Spark could prove that the other side already used the same partitioning rule.

The forced plan therefore reshuffles both inputs:

orders    → hash(customer_id) → shuffle bucket
customers → hash(customer_id) → same shuffle bucket

After the exchange, matching keys are colocated. Spark then sorts each side and uses SortMergeJoin to walk the sorted key ranges.

The extra operations expose the potential cost:

Exchange:
  Serialize and redistribute records.

Sort:
  Order each side by customer_id.

Shuffle Read/Write:
  Measure intermediate data moving between stages.

On a real cluster, this can involve network transfer, memory pressure, disk spill, serialization overhead, and task retries. This laptop run proves the plan shape, not a production latency or cost number.

Same rows, different physical work

Both executions returned:

joined_count=12
distinct_order_count=12

That gives us a useful separation:

Logical contract:
  Every valid order is enriched with its matching customer.

Physical strategy A:
  Broadcast the small customer side.

Physical strategy B:
  Shuffle both sides by customer_id and sort them.

A query plan change should not silently change the business meaning. The row count, order grain, and selected customer values are regression checks against that requirement.

How to inspect the join in the Spark UI

Run the forced version with the UI held open:

bash scripts/run_poc2.sh \
  --join \
  --disable-broadcast \
  --enable-ui \
  --keep-alive

Open the printed URL, normally:

http://localhost:4040

Inspect the tabs in this order:

SQL / DataFrame:
  Find SortMergeJoin and the two Exchange nodes.

Stages:
  Look for task counts and Shuffle Read/Write.

Jobs:
  See how the high-level join action becomes jobs and stages.

Then run the normal version separately:

bash scripts/run_poc2.sh \
  --join \
  --enable-ui \
  --keep-alive

Compare:

Normal:
  BroadcastExchange on customers
  BroadcastHashJoin
  no orders-side Exchange

Broadcast disabled:
  Exchange on both inputs
  Sort on both inputs
  SortMergeJoin

The UI summary may not show every source path or order row. Use the application output for row-level correctness and the SQL details for physical-plan evidence.

What to carry into production reasoning

A join strategy is not selected by the table names alone. Useful questions include:

Is the build side small enough to broadcast safely?
Is the join key unique on the dimension side?
What is the expected output grain?
How many rows can one key produce on each side?
Are the inputs already partitioned compatibly?
What is the shuffle volume?
Can the build side fit in executor memory?
What happens when estimates are wrong?
Is the result replay-safe and idempotent?
Does a tenant or security boundary require additional filtering before join?

Broadcasting a genuinely small dimension can avoid moving a very large fact table. Broadcasting a dimension that has grown beyond executor capacity can cause memory pressure or failure. Forcing a shuffle can be safer for very large relations, but it makes network, sort, spill, and skew behavior important.

What this POC has and has not covered

Verified in the current POC:

explicit schemas for both inputs;
inner-join cardinality on a clean many-to-one relationship;
customer-side broadcast selection;
forced hash partitioning by customer_id;
Exchange on both sides of a forced shuffle join;
SortMergeJoin and sorting after broadcast is disabled;
identical row-level results under both strategies;
eleven passing regression tests across the learning repository.

Not yet completed:

a deliberately duplicated customer key tested as a fixture;
a hot-key skew experiment;
malformed input and schema-boundary diagnosis;
write/replay and output idempotency behavior;
cluster-scale measurements.

Those omissions are deliberate. A plan can be understood before every failure mode has been implemented, but the POC should not claim those later guarantees yet.

Revisit checklist

What is the logical grain of the joined output?
Why do customer attributes repeat across multiple order rows?
What happens when a key is missing from an inner join?
What happens when a key is duplicated on the dimension side?
Why did Spark broadcast customers?
What does BuildRight mean?
What is the difference between BroadcastExchange and Exchange?
Why does a forced shuffle need both inputs to exchange?
What does hashpartitioning(customer_id, 200) mean?
Why is 200 not the same as 200 files or 200 workers?
Why does SortMergeJoin need sorted inputs?
Which values prove that the physical strategy change preserved correctness?
What would change if the customer table became too large to broadcast?

The next Spark exercise is skew: create a bounded hot key, observe one partition receive disproportionate work, and compare mitigation choices without hiding the underlying data-distribution problem.