Broadcast Spatial Joins with Apache Sedona

This guide gives you a complete PySpark and Apache Sedona recipe that performs a broadcast spatial join — a small reference layer is broadcast to every executor while the large layer stays partitioned — using ST_Intersects with a spatial index, plus a physical-plan check that proves the broadcast happened.

Context and prerequisites

A broadcast spatial join is the right pattern when one side of the join is small (a few thousand administrative or reference polygons, typically under ~100 MB) and the other is enormous. Instead of shuffling both sides across the network by a spatial partitioner, Sedona ships the small side to every executor and probes it with a local index — eliminating the large-side shuffle entirely. This recipe is the concrete companion to the distributed spatial compute with Apache Sedona topic area within the Spatial Query Engines & Compute Optimization section. You need Spark 3.5, apache-sedona 1.6.x matched to that Spark version, and a reference layer that genuinely fits in executor memory; if both sides are large, use the KDB-tree partitioned join from the parent topic instead.

Complete working solution

# pip install apache-sedona==1.6.1 pyspark==3.5.1
from sedona.spark import SedonaContext
from pyspark.sql.functions import broadcast, expr

# (1) Session with Sedona serializers registered
config = (
    SedonaContext.builder()
    .appName("sedona-broadcast-spatial-join")
    .config(
        "spark.jars.packages",
        "org.apache.sedona:sedona-spark-shaded-3.5_2.12:1.6.1,"
        "org.datasyslab:geotools-wrapper:1.6.1-28.2,"
        "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.9.0",
    )
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.kryo.registrator",
            "org.apache.sedona.core.serde.SedonaKryoRegistrator")
    .config("spark.sql.catalog.lake", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.lake.type", "rest")
    .config("spark.sql.catalog.lake.uri", "https://catalog.internal:8181")
    # let Sedona auto-broadcast small spatial sides up to 100 MB
    .config("spark.sql.autoBroadcastJoinThreshold", str(100 * 1024 * 1024))
    .getOrCreate()
)
sedona = SedonaContext.create(config)

# (2) Large partitioned layer: billions of pings from Iceberg
pings = sedona.sql("""
    SELECT device_id, event_ts, ST_GeomFromWKB(geom_wkb) AS geom
    FROM lake.telemetry.pings
    WHERE event_ts >= TIMESTAMP '2026-07-01 00:00:00'
""")

# (3) Small reference layer: a few thousand zone polygons from GeoParquet
zones = (
    sedona.read.format("geoparquet").load("s3a://ref/zones/")
    .selectExpr("zone_id", "geometry AS zone_geom")
)

# (4) Broadcast spatial join: hint the SMALL side; ST_Intersects drives it
joined = (
    pings.join(
        broadcast(zones),
        expr("ST_Intersects(geom, zone_geom)")
    )
    .select("device_id", "event_ts", "zone_id")
)

# (5) Materialize and persist back to Iceberg as WKB
(joined
   .withColumn("geom_present", expr("true"))
   .writeTo("lake.telemetry.pings_zoned")
   .using("iceberg")
   .createOrReplace())

print("joined rows:", joined.count())

Step-by-step walkthrough

1. Register Sedona on the session (block 1). The Kryo serializer and SedonaKryoRegistrator are mandatory — without them, geometry cannot be serialized for a broadcast and you get a serialization error at the first shuffle or collect. spark.sql.autoBroadcastJoinThreshold raised to 100 MB lets the optimizer choose a broadcast automatically for the small side; the explicit broadcast() hint in step 4 forces it regardless.

2. Read the large side (block 2). ST_GeomFromWKB reconstructs geometries from the Iceberg WKB column. The event_ts filter is pushed into Iceberg scan planning so the large side is already reduced before the join — the same predicate pushdown leverage described for the SQL engines.

3. Read the small reference side (block 3). Sedona reads GeoParquet natively and decodes its geometry column, so no ST_GeomFromWKB call is needed here. Aliasing to zone_geom avoids a name clash in the join.

4. Force the broadcast (block 4). Wrapping zones in broadcast() tells Spark to ship the entire small layer to every executor and build a local index of it there. Each large-side partition then probes that in-memory index with ST_Intersects — no shuffle of the billions of pings, which is the whole point. ST_Intersects is the load-bearing predicate; it returns true when the geometries share any point.

5. Write back as WKB (block 5). Persisting to Iceberg keeps the table engine-neutral so Trino and DuckDB can read it later.

Common errors and fixes

Verification

Prove that Spark chose a broadcast plan (not a shuffle join, not a Cartesian nested loop) and spot-check correctness:

# 1) The physical plan must contain a broadcast + spatial index probe
joined.explain()
# expect: "BroadcastSpatialJoin" (or Broadcast + RangeJoin) and
#         NO plain "SortMergeJoin" / full "BroadcastNestedLoopJoin"

# 2) Correctness: count for one known zone via an independent predicate
from pyspark.sql.functions import expr
one_zone = zones.where("zone_id = 'Z-0042'")
check = pings.join(
    broadcast(one_zone),
    expr("ST_Contains(zone_geom, geom)")
).count()
print("Z-0042 contained pings:", check)

The explain() output should show the small side under a broadcast exchange feeding a spatial join operator; if you instead see a SortMergeJoin, the broadcast hint did not apply and you should confirm the reference side’s estimated size.

To compare this against a shuffle-partitioned join, return to distributed spatial compute with Apache Sedona; for the SQL-engine equivalent see spatial joins across catalogs with Trino, and to prepare the Iceberg source see optimizing spatial joins with Iceberg Z-ordering. Canonical semantics for the join and broadcast hints are in the Apache Sedona SQL documentation.