#Broadcast vs Partitioned Spatial Joins in Sedona

Broadcast when the small side’s decoded size fits comfortably in executor memory; partition when it does not. The word doing the work is “decoded”: compressed GeoParquet expands three to six times when its geometry becomes JVM objects with an R-tree over them, so the file size that looks safe is routinely the one that produces an out-of-memory failure forty minutes into a job. This page is the sizing decision behind Apache Sedona for distributed spatial joins.

#Quick Reference

#Why Decoded Size Is the Number

A GeoParquet file at ZSTD level 3 stores geometry as compressed WKB. Broadcasting it means every executor holds, simultaneously: the decompressed WKB bytes, the parsed geometry objects, the R-tree nodes indexing them, and the attribute columns. Each of those is real memory and only the first bears any resemblance to the file size.

The expansion factor varies with geometry complexity. Simple points expand least — maybe two and a half times. Detailed polygons with hundreds of vertices expand most, because per-object overhead is paid per geometry and the R-tree adds a node per entry. Three to six times covers the range for typical vector layers, and the safe move is to measure rather than assume.

The consequence is that the interesting comparison is not “is this table small” but “does this table, decoded, fit in a fraction of one executor’s heap, with room left for the streaming side.”

#Deciding and Enforcing

# Requires: apache-sedona[spark]>=1.6, pyspark>=3.5  (Python 3.10+)
from __future__ import annotations

from dataclasses import dataclass

from pyspark.sql import DataFrame, SparkSession
from pyspark.sql import functions as F

# Measured expansion from compressed GeoParquet to decoded JVM objects plus
# an R-tree. Points sit near the bottom, dense polygons near the top.
EXPANSION = {"point": 2.5, "line": 4.0, "polygon": 6.0}
SAFE_HEAP_FRACTION = 0.33


@dataclass(frozen=True)
class JoinDecision:
    strategy: str
    decoded_gb: float
    budget_gb: float
    reason: str


def decide(
    compressed_gb: float,
    dominant_type: str,
    executor_heap_gb: float,
) -> JoinDecision:
    """Choose broadcast or partitioned from decoded size, not file size."""
    factor = EXPANSION.get(dominant_type.lower())
    if factor is None:
        raise ValueError(f"unknown geometry class {dominant_type!r}")
    if compressed_gb <= 0 or executor_heap_gb <= 0:
        raise ValueError("sizes must be positive")

    decoded = compressed_gb * factor
    budget = executor_heap_gb * SAFE_HEAP_FRACTION
    if decoded <= budget:
        return JoinDecision(
            "broadcast", decoded, budget,
            f"{decoded:.1f} GB decoded fits the {budget:.1f} GB broadcast budget",
        )
    return JoinDecision(
        "partitioned", decoded, budget,
        f"{decoded:.1f} GB decoded exceeds the {budget:.1f} GB budget",
    )


def join(
    spark: SparkSession,
    large: DataFrame,
    small: DataFrame,
    decision: JoinDecision,
    *,
    partitions: int = 512,
) -> DataFrame:
    """Apply the chosen strategy explicitly rather than hoping the planner agrees."""
    if decision.strategy == "broadcast":
        # Raising the threshold is not enough on its own: the planner's own size
        # estimate for a geometry column is unreliable, so hint the join too.
        spark.conf.set(
            "spark.sql.autoBroadcastJoinThreshold",
            str(int(decision.budget_gb * 1024**3)),
        )
        right = F.broadcast(small.alias("s"))
    else:
        spark.conf.set("sedona.join.gridtype", "kdbtree")
        spark.conf.set("sedona.join.numpartition", str(partitions))
        spark.conf.set("spark.sql.shuffle.partitions", str(partitions))
        right = small.alias("s")

    return large.alias("l").join(
        right, F.expr("ST_Intersects(l.geometry, s.geometry)"), how="inner"
    )


def assert_strategy(frame: DataFrame, expected: str) -> None:
    """Fail loudly if the planner ignored the decision."""
    plan = frame._jdf.queryExecution().executedPlan().toString()
    got = (
        "broadcast" if "BroadcastIndexJoin" in plan
        else "partitioned" if ("RangeJoin" in plan or "DistanceJoin" in plan)
        else "nested-loop"
    )
    if got != expected:
        raise RuntimeError(
            f"expected a {expected} join, planner chose {got}:\n{plan[:600]}"
        )

#Validation

Two numbers confirm the decision was right: peak executor memory, and shuffle bytes written. A broadcast join should show zero shuffle for the join stage; a partitioned join should show a peak well under the heap.

# From the Spark event log: shuffle written by the join stage, peak executor memory
spark-submit --conf spark.eventLog.enabled=true ... join_job.py
python3 - <<'PY'
import json
peak = shuffle = 0
for line in open("/tmp/spark-events/application_1"):
    event = json.loads(line)
    if event.get("Event") == "SparkListenerTaskEnd":
        metrics = event.get("Task Metrics", {})
        shuffle += metrics.get("Shuffle Write Metrics", {}).get("Shuffle Bytes Written", 0)
        peak = max(peak, metrics.get("Peak Execution Memory", 0))
print(f"shuffle written {shuffle/1e9:.2f} GB · peak execution memory {peak/1e9:.2f} GB")
PY

Expected results: a broadcast join writes essentially zero shuffle bytes for the join stage and peaks at roughly the decoded broadcast size plus the streaming batch; a partitioned join writes shuffle bytes on the order of both inputs combined and peaks much lower. A broadcast job peaking near the heap ceiling is one dataset growth away from failing.

#Edge Cases and Caveats

Broadcast repeated across many short queries. Each query rebuilds the index on every executor, so a workload of hundreds of small joins pays that cost hundreds of times. Cache the broadcast side across queries where the engine allows it, or accept a partitioned join whose per-query setup is cheaper.

A small side that grows. Boundaries get subdivided, footprints get re-surveyed, and a table that fitted last quarter does not this quarter. Bake the decision into the job as a computed threshold rather than a hard-coded flag, so growth changes the strategy automatically instead of producing an OOM.

A planner that ignores the hint. Adaptive query execution can override a broadcast hint when its runtime statistics disagree. The assert_strategy check catches this, and it is worth running in production rather than only in development — the nested-loop fallback is the failure it exists to prevent.

#Frequently Asked Questions

How large can the broadcast side of a spatial join be?

Judge it on the decoded size, not the file size. Compressed GeoParquet expands roughly three to six times when geometry becomes JVM objects and an R-tree is built over it, so a 2 GB file can occupy 10 GB per executor. A safe rule is to broadcast only when the decoded estimate fits within about a third of the executor heap, leaving room for the streaming side and the shuffle buffers.

Why does the planner choose the wrong join strategy so often?

Because its size estimates come from file statistics, and a geometry column’s compressed byte size is a poor predictor of its decoded footprint. The planner may see a 1.8 GB table and treat it as broadcastable when decoding will produce 9 GB, or refuse to broadcast a table that would have fitted comfortably. Setting the threshold explicitly, and hinting the join, replaces a guess with a measurement.

Is a broadcast join always faster when it fits?

Nearly always, because it eliminates the shuffle entirely — no data crosses the network between stages, and each executor queries a local R-tree. The exception is when the small side is close to the memory limit, where garbage collection pressure can make it slower than a clean partitioned join, and when the broadcast itself is repeated across many short queries, in which case building the index once per executor per query becomes the dominant cost.

#Related

• Apache Sedona for Distributed Spatial Joins — parent guide: why a spatial join is hard to distribute at all
• Tuning Spark Shuffle for Large Spatial Joins — what to do once the partitioned path is chosen
• Choosing Quadtree Depth for Uneven Point Density — the partitioner that keeps tasks balanced
• DuckDB Spatial Extension for GeoParquet — the single-node option when neither strategy is needed

← Back to Apache Sedona for Distributed Spatial Joins