Tuning Sedona Partitioning for Skewed Joins

This guide configures Sedona’s spatial partitioner for a large-versus-large join over unevenly distributed data, choosing the grid type and partition count from measurements rather than defaults, and verifying that the straggler is actually gone.

Context and prerequisites

When neither side of a spatial join fits in a broadcast, the partitioned join is the remaining strategy — and its performance is entirely determined by whether the partitioner divides the data evenly rather than the space evenly. This recipe uses Spark 3.5 with Sedona; the strategy choice is in choosing between broadcast and partitioned spatial joins, and the engine context in Sedona distributed spatial compute.

Grid type is the decision that matters

The tree partitioner builds its boundaries from a sample of the data, so its cells are small where the data is dense and large where it is sparse. That is exactly the property a uniform grid lacks and exactly the property a skewed spatial join needs.

The cost is a sampling pass before the join and a partitioner that is data-dependent rather than fixed — which means it must be rebuilt when the distribution changes, and that two runs over different data are not directly comparable.

Complete working solution

from pyspark.sql import SparkSession, functions as F
from sedona.spark import SedonaContext

spark = SedonaContext.create(SparkSession.builder
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.kryo.registrator", "org.apache.sedona.core.serde.SedonaKryoRegistrator")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .getOrCreate())

def choose_partition_count(spark, table: str, target_mb: int = 128) -> int:
    """Partition count from data volume, not from a default."""
    total = spark.sql(f"""
        SELECT sum(file_size_in_bytes) AS b FROM {table}.files
    """).collect()[0]["b"]
    return max(int(total / (target_mb * 1e6)), spark.sparkContext.defaultParallelism)

def configure(spark, parcels_table: str, buildings_table: str) -> None:
    n = max(choose_partition_count(spark, parcels_table),
            choose_partition_count(spark, buildings_table))
    spark.conf.set("sedona.join.gridtype", "kdbtree")   # adapts to the distribution
    spark.conf.set("sedona.join.numpartition", n)
    spark.conf.set("sedona.join.indexbuildside", "left")
    spark.conf.set("sedona.join.spatitionside", "left")
    spark.conf.set("spark.sql.shuffle.partitions", n)

def run_join(spark):
    return 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

Register the Kryo serialiser. Geometry serialises poorly through the default Java path, and on a partitioned join every feature crosses the network. This single configuration typically reduces shuffle volume by a factor of two to four.
Choose a tree-based grid type. A uniform grid divides space evenly, which for spatial data means dividing the workload unevenly. The tree variants sample the data and place boundaries where the density is.
Derive the partition count from volume. Aim for roughly 128 MB of input per partition, floored at the cluster’s default parallelism so small jobs still use the whole cluster. A fixed number tuned for one dataset is wrong for the next.
Enable adaptive skew handling as a backstop. Spark’s adaptive execution splits skewed shuffle partitions at runtime, which catches residual skew the partitioner did not anticipate. It is a safety net rather than a substitute for the partitioner.
Index the side with more features per partition. The tree index is built per partition on one side and probed by the other; building it on the denser side gives the probe more work to skip.

Common errors and fixes

Symptom	Cause	Fix
One task runs many times longer	Uniform grid on skewed data	Switch to a tree-based grid type
Shuffle volume enormous	Default Java serialisation	Register the Kryo serialiser
Too many tiny tasks	Partition count far above the data volume	Derive it from bytes, not from a habit
Duplicate result rows	Features replicated across partitions, not deduplicated	Deduplicate on the identifier pair
Job fails on executor memory	Index built on the side with huge geometries	Build the index on the other side; check vertex counts

Verifying the fix

def assert_no_straggler(spark, stage_id: int, max_ratio: float = 3.0):
    info = spark.sparkContext.statusTracker().getStageInfo(stage_id)
    durations = sorted(t.duration for t in info.taskInfos)
    median = durations[len(durations) // 2]
    ratio = durations[-1] / median
    assert ratio <= max_ratio, f"straggler remains: slowest task {ratio:.1f}× median"

The ratio is the acceptance test, and it is more informative than the total runtime because it separates two questions: whether the work is balanced, and whether there is too much of it. A job with a good ratio that is still slow needs more resources; one with a poor ratio needs a better partitioner, and adding resources will not help.

Record the ratio with each run. Skew re-emerges as the data grows, and a recorded series turns the eventual re-tuning into a scheduled change rather than a surprise.

Duplication at Partition Boundaries

A spatially partitioned join replicates any feature that spans a partition boundary into every partition it touches, and that replication has two consequences worth handling deliberately.

The first consequence sets an upper bound on how fine the partitioning should be: past a certain point, extra partitions replicate more features than they balance, and the shuffle grows faster than the parallelism helps. Measuring shuffle bytes against input bytes reveals it — a ratio much above two means the partitioner is dividing more finely than the geometry sizes justify.

The second consequence is a correctness issue. A pair of features sharing several partitions matches in each, so the join emits duplicates. Deduplicating on the identifier pair is correct and adds a shuffle; assigning each pair to a canonical partition and filtering to it avoids the shuffle and is worth the extra clause on large joins.

Neither problem appears in a broadcast join, which is one more reason to reduce the smaller side and broadcast where it is at all possible.

When the Partitioner Is Not the Problem

Three situations produce a straggler that no partitioner setting fixes, and recognising them saves a lot of fruitless tuning.

One enormous geometry. A single feature with hundreds of thousands of vertices costs more to evaluate than a hundred thousand simple ones, and it lives in one partition by definition. The remedy is to simplify it for join purposes or to split it into pieces sharing an identifier — a data change rather than a configuration one.

A genuinely dense region. Where a metropolitan area really does contain a third of the data and the join is many-to-many within it, the work is irreducibly concentrated. Adaptive execution can split the shuffle partition, but the candidate pairs remain, and the honest answer is that this partition takes longer.

A skewed result rather than a skewed input. Both inputs may be balanced while the output concentrates — a flood zone overlapping every parcel in a district produces a fan-out that no input partitioning anticipates. Here the fix is in the query: aggregate earlier, or restrict the join to the pairs the analysis actually needs.

The diagnostic that distinguishes these from ordinary partitioner skew is to compare the input rows and the output rows per partition. Input skew is a partitioner problem; output skew is a query problem; and a partition with balanced input, balanced output and a long duration contains a pathological geometry.

A Tuning Sequence That Converges

Working through the settings in a fixed order avoids the common experience of changing several things at once and being unable to attribute the improvement.

Start by registering Kryo and measuring. It is a one-line change with a large effect on shuffle volume and it interacts with nothing else, so its contribution is clean.

Then switch the grid type to a tree-based partitioner and measure the task-duration ratio. This is the change that addresses skew directly, and if the ratio does not improve, the skew is one of the three non-partitioner cases above and further partitioner tuning is wasted.

Then set the partition count from the data volume and measure both the ratio and the shuffle bytes. Too few partitions leaves large tasks; too many inflates replication. The measurement distinguishes them and one adjustment usually suffices.

Finally enable adaptive skew handling as a backstop and confirm it is not doing much — if it is splitting many partitions, the partitioner is still not fitting the data and the previous step needs revisiting.

Record all four measurements at each step, and record the cluster shape alongside them. A tuning result is only meaningful for the cluster it was measured on, and a configuration inherited from a differently-sized cluster is a common source of settings nobody can explain. Recording the cluster shape alongside the settings is what lets a future reader tell an inherited configuration from a measured one. The distinction matters because an inherited configuration is safe to change and a measured one is not. Write down which it is, next to the setting, and the next tuning session starts from knowledge rather than from guesswork. A settings file with reasons is worth several with better values. Reasons outlive values.

Tuning Sedona Partitioning for Skewed Joins

Related in this section