Distributed Spatial Compute with Apache Sedona

Apache Sedona (formerly GeoSpark) extends Spark 3.5 with a distributed spatial type system, spatial partitioners, and index-backed join operators, making it the tool of choice when a spatial join is too large for any single machine. Where a single-node engine loads one dataset into memory, Sedona shards billions of geometries across a Spark cluster, builds a distributed spatial index (KDB-tree or quad-tree), and executes range and join queries in parallel. This topic area covers the SpatialRDD and Sedona SQL programming models, spatial partitioning and index construction, reading and writing both Apache Iceberg and GeoParquet from Sedona, and the concrete threshold at which distribution beats single-node DuckDB geospatial analytics. It belongs to the Spatial Query Engines & Compute Optimization section and complements the SQL-federation approach in Trino spatial SQL and cross-catalog federation.

When to use this

Sedona earns its operational overhead only when the data genuinely exceeds single-node capacity or when the spatial join is quadratic and both sides are large. Below roughly 50–100 GB of geometry, a single-node engine will almost always finish faster because it skips job scheduling and shuffle. The decision is about data size, join cardinality, and whether the output feeds a heavier Spark transformation DAG.

If your large-large spatial join OOMs or runs for hours on a single node, that is the signal to move to Sedona’s partitioned, index-backed join.

Prerequisites and environment setup

Pin Spark 3.5 and a matching Sedona release. Sedona ships as Scala/Java jars plus the apache-sedona Python package; the two must agree on Spark and Scala versions. Register the Sedona SQL functions and serializers on the session, and add the Iceberg and Sedona jars to the classpath.

# pip install apache-sedona==1.6.1 pyspark==3.5.1
from sedona.spark import SedonaContext

config = (
    SedonaContext.builder()
    .appName("sedona-spatial-join")
    # Sedona + Iceberg runtime jars (match Spark 3.5 / Scala 2.12)
    .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",
    )
    # Sedona geometry serializer (Kryo) — required for shuffle
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.kryo.registrator", "org.apache.sedona.core.serde.SedonaKryoRegistrator")
    # Iceberg REST catalog
    .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")
    .getOrCreate()
)
sedona = SedonaContext.create(config)

Verify registration with sedona.sql("SELECT ST_Point(0.0, 0.0)").show(). A Undefined function ST_Point error means the serializer/registrator config did not take — the jars and SedonaContext.create step are both required.

Step-by-step implementation

1. Read spatial data from Iceberg and GeoParquet

Sedona reads GeoParquet natively and reads Iceberg through the standard Spark catalog, then reconstructs geometries with ST_GeomFromWKB. Keep everything in EPSG:4326 lon/lat so downstream predicates are unambiguous.

# Large fact layer from Iceberg (WKB in a binary column)
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'
""")

# Reference layer from GeoParquet (geometry column decoded automatically)
zones = sedona.read.format("geoparquet").load("s3a://ref/zones/")
zones.createOrReplaceTempView("zones")
pings.createOrReplaceTempView("pings")

2. Let Sedona build the spatial partitioning and index

The Sedona SQL optimizer recognizes an ST_ predicate in the join condition and injects a distributed spatial join: it partitions both inputs with a KDB-tree (equal-load, skew-aware) and builds a local R-tree per partition. You enable the range-join optimization and set the partition count; you do not hand-write the partitioner.

sedona.conf.set("sedona.join.numpartition", "200")
sedona.conf.set("sedona.join.gridtype", "kdbtree")   # or "quadtree"
sedona.conf.set("sedona.join.indextype", "rtree")

result = sedona.sql("""
    SELECT p.device_id, z.zone_id, p.event_ts
    FROM pings p JOIN zones z
      ON ST_Intersects(p.geom, z.geometry)
""")
result.cache()

For the small-reference-layer case where a broadcast is cheaper than a shuffle, use the explicit broadcast hint pattern documented in broadcast spatial joins with Apache Sedona.

3. Write results back to Iceberg

Encode the geometry back to WKB before writing so the Iceberg schema stays engine-neutral and readable by Trino and DuckDB (the encoding contract is covered under Iceberg spatial type support).

from pyspark.sql.functions import expr

(result
   .withColumn("geom_wkb", expr("ST_AsBinary(geom)"))
   .drop("geom")
   .writeTo("lake.telemetry.pings_zoned")
   .using("iceberg")
   .createOrReplace())

Verification and testing

Confirm the optimizer actually chose the distributed spatial join rather than a Cartesian product by inspecting the physical plan; a healthy plan contains a RangeJoin (or DistanceJoin) node, not BroadcastNestedLoopJoin over the full product.

result.explain()   # look for "RangeJoin" and the spatial partitioner
print("rows:", result.count())

# bbox sanity: joined pings must fall within the union bbox of matched zones
result.selectExpr(
    "min(ST_XMin(geom)) minx", "min(ST_YMin(geom)) miny",
    "max(ST_XMax(geom)) maxx", "max(ST_YMax(geom)) maxy"
).show()

Performance and tuning

Sedona performance is dominated by partition balance and index construction cost. Concrete knobs and ranges:

• sedona.join.numpartition: target 2–4 partitions per executor core; too few starves parallelism, too many inflates index build overhead. For a 200-core cluster, 400–800 is a reasonable band.
• sedona.join.gridtype: use kdbtree for skewed data (dense cities, sparse ocean) because it equalizes load; quadtree is fine for uniform distributions and builds faster.
• spark.sql.autoBroadcastJoinThreshold: raise it (or use an explicit hint) when the reference side is under ~100 MB so Sedona broadcasts instead of shuffling.
• spark.executor.memory / spark.memory.fraction: geometry objects and R-tree nodes are heap-heavy; budget 8–16 GB per executor for 100M+ geometry joins and enable off-heap if GC pauses dominate.

At the crossover point, a distributed join over ~500 GB with balanced KDB-tree partitions typically runs 3–8x faster than a single node that has to spill; below ~50 GB the single node wins because Sedona’s job-startup and shuffle costs are not amortized. Pre-sorting the Iceberg source with Z-ordering cuts the bytes read before partitioning even begins.

Common errors and fixes

For authoritative API and configuration reference, consult the Apache Sedona documentation and the Sedona spatial join tuning guide. To decide empirically whether Sedona, Trino, or DuckDB fits a given workload, run the harness in benchmarking spatial query engines on GeoParquet.