Geospatial Orchestration Hub

Handling Skewed Partitions in Spatial Workloads

A national run finishes when its last partition finishes, and its last partition is always a city. Spatial data is distributed like the population it describes, so any scheme that divides the world evenly divides the work extremely unevenly — routinely a hundred to a thousand times between the median cell and the worst. The fix is not a better tessellation but a weight computed before scheduling, used to subdivide the heavy partitions and to schedule the rest longest-first.

When to Use This Pattern

  • A run’s duration is set by a handful of partitions while most workers finish early and idle.
  • Timeouts fire on the same cells every time, which is skew being reported as failure.
  • Adding workers does not make the run faster, the clearest symptom that one unit is the critical path.
  • Partition weights vary by more than about twenty times, which is where the effect becomes worth engineering around.

Complete Working Example

Weigh, split, then schedule heaviest-first.

from __future__ import annotations

from dataclasses import dataclass, replace

from prefect import flow, task, get_run_logger


@dataclass(frozen=True)
class Unit:
    key: str
    weight: float          # features, pixels, or an estimate of either
    level: int = 0         # how many times this unit has been subdivided


@task(timeout_seconds=600)
def weigh(keys: list[str]) -> list[Unit]:
    """Metadata only — a count query or a header read, never the work itself."""
    counts = feature_counts_by_partition(keys)      # one grouped query
    return [Unit(k, float(counts.get(k, 0))) for k in keys]


def split(units: list[Unit], target: float, max_level: int = 3) -> list[Unit]:
    """Subdivide anything above the target until it fits or we run out of levels."""
    out: list[Unit] = []
    for u in units:
        if u.weight <= target or u.level >= max_level:
            out.append(u)
            continue
        children = subdivide(u.key)                  # 4 tiles, or 7 h3 children
        share = u.weight / len(children)
        out.extend(Unit(c, share, u.level + 1) for c in children)
    # One more pass: a child of a dense parent can still be too heavy.
    return out if all(u.weight <= target or u.level >= max_level for u in out) \
        else split(out, target, max_level)


@task(retries=2, timeout_seconds=1800, tags=["unit"])
def process(unit: Unit) -> str:
    return do_work(unit.key)


@flow(name="balanced-run")
def run(keys: list[str], target_weight: float = 2_000) -> int:
    units = split(weigh(keys), target_weight)

    # Longest-first: the classic scheduling result. Submitting the heavy units
    # last leaves one worker finishing alone while the others have nothing to do.
    ordered = sorted(units, key=lambda u: -u.weight)

    log = get_run_logger()
    log.info("units: %d after splitting (was %d), heaviest %.0f, median %.0f",
             len(ordered), len(keys), ordered[0].weight,
             ordered[len(ordered) // 2].weight)

    states = process.map(ordered, return_state=True)
    return sum(s.is_completed() for s in states)
What the tail costs, and what removing it buysWith one heavy partition the run takes fifty-two minutes while most workers idle. After splitting and longest-first scheduling the same work takes fourteen.8 workers, 412 partitionsbeforeone city cell — 52 min, and the run is not finished until it isidle from minute 12afterthe city split into 16 units, spread across the fleet — 14 minSame total work, same workers. The only change is that the largest unit is now the size of the others.
Adding workers to the top case changes nothing, because the critical path is one indivisible unit rather than a shortage of capacity.

The weighing step must use metadata rather than the work itself, or it is not a saving. A grouped count query against an indexed partition column, or a header read per raster, gives a good weight in seconds for the whole set; actually processing a unit to find out how long it takes is simply doing the work twice. Where no cheap proxy exists, the previous run’s durations are an excellent weight — the distribution changes slowly, so last night’s timings predict tonight’s well enough to schedule by.

Parameter & Option Reference

Setting Typical Spatial notes
Weight source count query, header read, last run’s timings Must be far cheaper than the work; otherwise it is not a saving.
Target weight ~2× the median Aggressive enough to flatten the tail, loose enough not to shatter the set.
max_level 3 Bounds the recursion. A cell that is still heavy after three splits is a genuine hotspot.
Scheduling order descending weight Longest-processing-time-first; a well-known and very cheap improvement.
Split factor 4 (tiles) or 7 (H3) The scheme’s own children, so the keys stay canonical.
Superlinear work square the weight Overlays and unions cost more than linearly in feature count.
Route instead of split for indivisible units A single huge scene goes to a larger worker rather than being subdivided.
Splitting until the tail is goneA partition weighing twenty-eight thousand is split into four children, and the two heaviest of those are split again, until every unit is near the target of two thousand.28 640level 09 100level 17 400level 12 3001 9002 1001 750target 2 00016 units from 1skew 698x to 3x
The recursion is bounded by level rather than by weight, so a genuinely extreme hotspot produces a large unit rather than an infinite descent.

Verification & Testing

def test_splitting_reduces_skew() -> None:
    units = [Unit(f"c{i}", w) for i, w in enumerate(REAL_WEIGHTS)]
    after = split(units, target=2_000)
    before_skew = max(u.weight for u in units) / median(u.weight for u in units)
    after_skew = max(u.weight for u in after) / median(u.weight for u in after)
    assert after_skew < before_skew / 10


def test_splitting_preserves_total_weight() -> None:
    units = [Unit(f"c{i}", w) for i, w in enumerate(REAL_WEIGHTS)]
    after = split(units, target=2_000)
    assert sum(u.weight for u in after) == pytest.approx(sum(u.weight for u in units))


def test_splitting_terminates_on_a_single_hotspot() -> None:
    after = split([Unit("dense", 10_000_000)], target=100, max_level=3)
    assert all(u.level <= 3 for u in after)


def test_children_are_canonical_keys() -> None:
    for child in subdivide("12/2147/1398"):
        assert Tile.from_path(child)          # raises if the key is not canonical


def test_scheduling_is_longest_first(monkeypatch) -> None:
    submitted: list[float] = []
    monkeypatch.setattr("mymodule.process", recording(submitted))
    run(KEYS)
    assert submitted == sorted(submitted, reverse=True)

The weight-preservation test is worth keeping because a split that loses or duplicates weight is a split that loses or duplicates work, and the symptom is a coverage figure that quietly stops adding up. Distributing a parent’s weight equally among its children is an approximation — a city cell’s density is not uniform across its four quarters — but it must be an approximation that conserves the total, or every downstream estimate built on the weights inherits the error.

Ordering alone, before any splittingWith arbitrary ordering the fleet drains gradually and finishes with one worker running alone. Scheduling the heaviest units first keeps the fleet busy until near the end.busyarbitrary orderlongest-firststartendOne line of sorting, typically 15 to 30 per cent off the wall clock, and no change to the work at all.
Worth doing even when splitting is impractical, because it costs nothing and needs no knowledge of what the units contain.

A question worth settling early is where the splitting belongs — in the flow, as above, or in the partition scheme itself. Doing it in the flow keeps the scheme simple and the splitting adaptive: last night’s dense cell can be split tonight and left whole next week, without any change to how partitions are named. Baking it into the scheme, by declaring the dense regions at a finer resolution permanently, gives a stable partition set that ledgers and caches agree with across runs. The first is better when density moves; the second is better when it does not, and for most infrastructure data — buildings, roads, parcels — density moves very slowly, which argues for eventually promoting a repeatedly-split region into the scheme itself.

There is also a limit worth respecting. Splitting helps when a unit’s work divides cleanly across its children, which is true for rendering, warping and per-feature analysis and false for anything computing a global property of the partition — a network topology, a watershed, a connected-component labelling. For those, subdividing produces four wrong answers instead of one right one, and the correct response is routing rather than splitting. Knowing which kind of operation a task performs is what stops a balancing measure from becoming a correctness bug.

Common Pitfalls

  • Discovering skew as a timeout. By then the run is over budget; weigh before scheduling instead.
  • Weighing by doing the work. A weight must come from metadata or from history, or it doubles the cost.
  • Unbounded recursion. A genuinely extreme hotspot will subdivide forever unless a level cap stops it.
  • Splitting into non-canonical keys. The children must be the scheme’s own children, or ledgers and caches stop matching.
  • Ignoring superlinear cost. For overlays and unions, feature count underestimates the heavy cells badly; square the weight.
  • Adding workers. If one unit is the critical path, more workers idle earlier and the run takes exactly as long.

Frequently Asked Questions

What if the heavy unit cannot be split?

Route it instead: send it to a work pool with more memory and a longer timeout, and start it first. A single enormous scene is indivisible without changing the operation, and routing gets most of the benefit; see right-sizing workers for raster mosaic jobs.

Is longest-first worth doing on its own?

Yes, and it is the cheapest improvement available — one sorted call, no knowledge of the data, typically fifteen to thirty per cent off the wall clock. It is the first thing to try when a run has visible idle time near the end.

How do I choose the target weight?

Roughly twice the median, which flattens the tail without shattering the partition set into thousands of tiny units. Then check the resulting count: if it has grown by more than about half, the target is too aggressive and orchestration overhead will eat the gain.

Does this interact with caching?

Carefully. A split changes the unit’s key, so a subdivided partition’s cached results are keyed differently from its parent’s. Keeping the split deterministic — the same weights produce the same children — means the cache still works run to run, which is why the weight source should be stable rather than sampled.

What about skew across time rather than space?

The same treatment with a different weight: a delivery day carrying ten times the usual volume is a heavy partition in the temporal dimension. Weighing by expected volume and splitting the date range works exactly as it does spatially.

← Partitioning Strategies for Spatial Workloads

Related in this section