Broadcast joins and build-side limits

Last updated: 6 Oct 20265 min read

tutorial

A broadcast join copies a bounded lookup relation to workers so large fact partitions can join locally without shuffling the fact side.

Choose by measured bytes

A merchant lookup with 240 records may be small in source storage but much larger after deserialization or expansion. Measure the encoded and in-memory size, then enforce a cap before broadcasting. A join hint is a request to the planner, not proof that the chosen plan fits executor memory. Inspect the executed plan and task memory profile.

Check cardinality first

One lookup row per merchant key is required when each order line should remain one row. If the lookup contains two active records for the same merchant, broadcasting makes the wrong multiplication faster. Reject duplicate business keys or first resolve a temporal version using an as-of rule. Check source and result row counts.

Account for missing keys

A left join should preserve every fact, including a merchant not present in the lookup. Represent the unresolved member explicitly rather than silently dropping that order. A later lookup correction may require a targeted repair. Late dimension repair owns that lifecycle.

Understand the cost switch

Broadcasting B bytes to W workers costs roughly B × W network and memory exposure. Shuffling a large fact of F bytes may cost far more, but broadcasting a growing table can cause driver or executor failure. A sampled byte estimate, a safety margin and a fallback shuffle strategy are more reliable than a fixed row-count threshold.

Test under skew

Broadcast can remove fact-side movement for the join, but a downstream group-by may still concentrate one account on a reducer. Treat the join and aggregate as separate plan stages. Compare output keys, unmatched facts, shuffle bytes and p95 task duration before publishing the change.

Implementation

merchant_rows = [
    {"merchant_id": "m-47", "region": "north"},
    {"merchant_id": "m-83", "region": "west"},
]
order_lines = [
    {"line_id": "line-47", "merchant_id": "m-47", "amount_cents": 7300},
    {"line_id": "line-48", "merchant_id": "m-99", "amount_cents": 1200},
]

def enrich_orders(facts, lookup_rows):
    lookup = {row["merchant_id"]: row["region"] for row in lookup_rows}
    if len(lookup) != len(lookup_rows):
        raise ValueError("merchant lookup key is not unique")
    return [{**fact, "region": lookup.get(fact["merchant_id"], "unknown")}
            for fact in facts]

enriched = enrich_orders(order_lines, merchant_rows)
assert len(enriched) == len(order_lines)
assert enriched[1]["region"] == "unknown"

Performance and operating cost

Building the lookup takes O(B) time and space for B lookup rows; probing F fact rows takes O(F) expected time locally. In a cluster, each worker retains its own copy, so aggregate memory approaches W × B. Missing-key and duplicate-key checks protect correctness independently of the planner choice.

Common Mistakes

Do not broadcast from a row-count guess without a byte and memory cap.
Do not let duplicate lookup keys multiply financial facts.
Do not claim the entire job is shuffle-free because one join is local.