Choosing Between Broadcast and Partitioned Spatial Joins

This guide gives a measurement-driven decision procedure for the two distributed spatial join strategies, with the three numbers that decide it, the code to compute them, and the plan checks that confirm the engine did what you asked.

Context and prerequisites

The choice is usually made by habit and is usually wrong in one direction: teams either broadcast a side that has grown too large, or run a partitioned join for a case a broadcast would have handled with no shuffle at all. This recipe runs on Spark 3.5 with Sedona and Iceberg 1.4; the anatomy behind the strategies is in spatial join optimization, and the Sedona-side mechanics in broadcast spatial joins with Apache Sedona.

The decision, in one diagram

The middle branch is the one most often skipped, and it is where most production joins actually belong. A reference layer of a few gigabytes is rarely a few gigabytes of information — it is a few gigabytes of cartographic vertex detail that the join does not use. Simplifying it to the join’s resolution, or restricting it to the extent of the other side, routinely brings it under the broadcast threshold and removes the shuffle entirely.

Complete working solution

from pyspark.sql import functions as F

def measure_sides(spark, small_table: str, large_table: str, key: str) -> dict:
    small = spark.table(small_table)
    small_mb = (small.select(F.sum(F.length("geom_wkb")).alias("b"))
                     .collect()[0]["b"] or 0) / 1e6

    # Vertex detail is what makes a small reference layer large.
    detail = small.selectExpr("percentile_approx(ST_NPoints(ST_GeomFromWKB(geom_wkb)), 0.5) p50",
                              "percentile_approx(ST_NPoints(ST_GeomFromWKB(geom_wkb)), 0.99) p99")
    d = detail.collect()[0]

    # Skew of the join key on the large side, from a 1% sample.
    skew = (spark.table(large_table).sample(0.01)
                 .groupBy(key).count()
                 .selectExpr("max(count) mx",
                             "percentile_approx(count, 0.5) med")
                 .collect()[0])
    key_skew = (skew["mx"] / skew["med"]) if skew["med"] else float("inf")

    if small_mb < 200:
        strategy = "broadcast"
    elif small_mb < 2000:
        strategy = "reduce_then_broadcast"
    else:
        strategy = "partitioned"

    return {"small_side_mb": small_mb, "median_vertices": d["p50"],
            "p99_vertices": d["p99"], "key_skew": key_skew,
            "strategy": strategy,
            "warning": ("key skew will produce a straggler in a partitioned join"
                        if strategy == "partitioned" and key_skew > 4 else None)}

# Strategy A — broadcast, after reducing the small side if needed.
regions = spark.table("reference.regions")
if measurements["strategy"] == "reduce_then_broadcast":
    regions = regions.selectExpr(
        "region_id",
        "ST_AsBinary(ST_SimplifyPreserveTopology(ST_GeomFromWKB(geom_wkb), 0.0001)) AS geom_wkb")

result = spark.sql("""
SELECT /*+ BROADCAST(r) */ t.asset_id, r.region_id
FROM   lakehouse.spatial.telemetry t
JOIN   regions r
  ON   ST_Intersects(ST_GeomFromWKB(t.geom_wkb), ST_GeomFromWKB(r.geom_wkb))
""")

# Strategy B — spatially partitioned, for genuinely large-versus-large.
spark.conf.set("sedona.join.gridtype", "kdbtree")
spark.conf.set("sedona.join.numpartition", 512)

result = spark.sql("""
SELECT p.parcel_id, b.building_id
FROM   lakehouse.spatial.parcels p
JOIN   lakehouse.spatial.buildings b
  ON   ST_Intersects(ST_GeomFromWKB(p.geom_wkb), ST_GeomFromWKB(b.geom_wkb))
""")

Step-by-step walkthrough

Measure serialised bytes, not rows. A thousand detailed coastlines and a million points can differ by two orders of magnitude in memory while the row counts suggest the opposite. sum(length(geom_wkb)) is the number that predicts whether a broadcast fits.
Measure vertex detail separately. A high median vertex count is the signal that the middle branch applies — the layer is large because of detail rather than because of content, and simplification will reduce it dramatically without changing the join result at the resolution being used.
Measure key skew before choosing the partitioned strategy. A partitioned join over a skewed key inherits the skew as a straggler, and its advantage over a reduced broadcast can vanish entirely. Where the skew is above about four, fixing the layout comes first.
Simplify to the join’s tolerance, not to a round number. The tolerance should be well below the accuracy the join’s answer needs — typically a metre for administrative assignment — so the result is unchanged for every point that is not within a metre of a boundary.
Let the partitioner choose its grid type. A tree-based partitioner adapts to the data’s distribution, which is precisely what a uniform grid fails to do on spatial data. Fixing the partition count is more useful than fixing the grid type.

Common errors and fixes

Symptom	Cause	Fix
Broadcast hint ignored, shuffle appears	Estimated side size exceeded the threshold	Raise `autoBroadcastJoinThreshold` after measuring, or reduce the side
Executors run out of memory during broadcast	Side larger in memory than serialised	Measure in-memory footprint too; geometry objects inflate substantially
Partitioned join has one very slow task	Key skew	Apply adaptive resolution or salt the hot cells before joining
Both strategies are slow	The scan is the problem, not the join	Check bytes read first; layout precedes join tuning
Result counts differ between strategies	Duplicate matches at cell borders not deduplicated	Deduplicate on the identifier pair in the partitioned path

Verification

def assert_strategy(df, expected: str):
    plan = df.queryExecution.executedPlan.toString()
    exchanges = plan.count("Exchange")
    if expected == "broadcast":
        assert "BroadcastExchange" in plan, "the hint was declined"
        assert exchanges <= 1, f"unexpected shuffle: {exchanges} exchanges"
    else:
        assert exchanges == 1, f"expected one shuffle, found {exchanges}"

Run the assertion in the job rather than checking it by hand. A broadcast that silently degrades to a shuffle — because the reference layer grew past the threshold over six months — is the single most common cause of a spatial job that used to finish in four minutes and now takes two hours, and nothing in the output indicates it.

Record the measured side size alongside the result each run. The trend is what predicts the day the strategy needs revisiting, and having it recorded turns that from a surprise into a scheduled change.

The Middle Branch in Practice

Reducing a side to make it broadcastable is the highest-leverage move available here, and it has three variants that apply in different situations.

The extent pre-filter is first because it is free and reversible: computing the large side’s bounding box is one aggregate, and restricting the reference layer to it removes everything that could not possibly match. On a job scoped to one region against a national reference layer this alone is often sufficient, and it introduces no accuracy question at all.

Simplification comes second because it does have an accuracy cost, and that cost should be quantified with a disagreement rate rather than accepted implicitly — the technique is covered in simplifying geometries for analytical layers.

Projection is last because it is small, but it is genuinely free and frequently overlooked: broadcasting a reference table with forty attribute columns ships thirty-nine of them to every executor for no purpose. Join on identifier and geometry, and bring the attributes back with a second, ordinary join on the much smaller result.

When the Partitioned Join Is Genuinely Required

Three situations leave no reduction available, and for those the partitioned strategy is correct rather than a fallback.

Both sides are facts. A join between two large observation datasets — vehicle traces against traffic sensor readings, say — has no small reference side to reduce. Neither side is a lookup, both grow continuously, and the extent filter helps only if the job is regionally scoped.

The reference layer is genuinely detailed and the detail matters. A parcel-level join for a legal determination cannot be simplified, and the parcels of a metropolitan area are large in every representation. Here the accuracy requirement removes the middle branch by definition.

The join is many-to-many with high fan-out. Overlapping polygon layers — flood zones against land parcels — produce results larger than either input, and a broadcast does not help because the cost is in the candidate evaluation rather than the data movement.

In all three, the work shifts to controlling skew, because a partitioned join over spatial data will be skewed unless the partitioner adapts. Use a tree-based partitioner rather than a uniform grid, set the partition count from the data volume rather than accepting a default, and check the task duration spread after every run — a factor above three means one partition is doing the work of several and the remedy is in the layout rather than in the cluster.

The measurement to record here is shuffle bytes against input bytes. A healthy partitioned join shuffles roughly the size of its inputs once; one shuffling several times that is duplicating features across too many partitions, which points at a partitioner resolution that is too fine for the geometry sizes involved. Recording it each run gives the trend that says when the partition count needs raising, which otherwise only becomes apparent when a job starts spilling. The trend also settles arguments about whether a job “used to be faster”, which are otherwise unresolvable and consume a surprising amount of time. For the layout work that makes either strategy affordable in the first place, see spatial partitioning schemes — a join tuned against an unpartitioned table is tuning the wrong layer. Layout first, strategy second, cluster size last — in that order the work compounds rather than competing. Reversing it produces a larger cluster running the same badly-shaped join.

Choosing Between Broadcast and Partitioned Spatial Joins

Related in this section