Spatial Epi Public Health GIS

Making a Spatial Pipeline Idempotent and Restartable

This guide, part of Orchestrating Spatial Surveillance Pipelines, takes the property the parent guide asks for and examines the places it quietly fails in real spatial pipelines.

Problem Context & Constraints

Idempotency means running a stage twice leaves the system in the same state as running it once. It is the property that makes restart mechanical, makes an accidental double-trigger harmless, and makes a backfill predictable. Most stages are idempotent by accident when they are written as a read-transform-write of a whole output file, and most lose it as soon as someone optimises.

Five operations break it, all of them common in spatial work.

Appending to a shared output. Writing new rows into an existing table rather than replacing a partition means a second run doubles them. This is the most frequent cause and the easiest to introduce while making a stage “incremental”.

Unseeded randomness. Geomasking, differential privacy noise, bootstrap resampling and any sampling step produce different output on a second run unless the seed is fixed and derived from the run context rather than from the clock.

External service calls with side effects. A geocoder that charges per call, a routing service with a quota, a notification that emails a partner — none of these are undone by a restart, and a naive re-run repeats them.

Order-dependent results. Spatial joins that break ties by input order, nearest-neighbour queries with equidistant candidates, and dissolve operations that depend on row order all produce different output from the same data in a different order.

Accumulating auxiliary state. A geocode cache that is written during the stage, a spatial index persisted between runs, a deduplication table — each is a store whose contents differ on the second run, which can change the result even when the main inputs did not.

Five Breakers and Their Repairs Five paired rows. Appending to a shared table is repaired by replacing a whole partition instead. Unseeded randomness is repaired by deriving the seed from the run identifier and stage name. External calls with side effects are repaired by an idempotency key that the service or a local ledger deduplicates on. Order-dependent results are repaired by an explicit deterministic sort before any operation whose output depends on order. Accumulating auxiliary state is repaired by treating the cache as an input whose version is recorded, not as mutable state. Each breaker has one repair, and none of them costs much breaks idempotency repair append to a shared table replace a whole partition unseeded randomness seed from run id and stage name external call with a side effect idempotency key plus a local ledger order-dependent tie-breaking explicit sort on a stable key first accumulating auxiliary state treat the cache as a versioned input The repairs are cheap; discovering the breakage during an incident is not

Prerequisites

  • python 3.11
  • pandas 2.2.2, geopandas 0.14.4, numpy 1.26.4
  • The run context and atomic write helpers from the parent guide
  • A storage layout that supports replacing a partition atomically
  • A ledger table for external calls, keyed by an idempotency key

Step-by-Step Solution

Apply the five repairs. Each is a small change to how a stage is written, and together they make restart a mechanical operation.

# Idempotency helpers: partition replace, derived seeds, call ledger, stable ordering.
# Pinned: pandas==2.2.2, geopandas==0.14.4, numpy==1.26.4
import hashlib
import logging
import os
import shutil

import numpy as np
import pandas as pd
import geopandas as gpd

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("pipeline.idempotent")


def replace_partition(df: pd.DataFrame, partition_dir: str) -> None:
    """Write a partition by building it aside and swapping — never by appending."""
    staging = partition_dir + ".staging"
    if os.path.exists(staging):
        shutil.rmtree(staging)
    os.makedirs(staging, exist_ok=True)
    df.to_parquet(os.path.join(staging, "part-0.parquet"), index=False)
    if os.path.exists(partition_dir):
        shutil.rmtree(partition_dir)      # previous contents are fully discarded
    os.replace(staging, partition_dir)
    log.info("partition replaced: %s rows=%d", partition_dir, len(df))


def derived_seed(run_id: str, stage: str, extra: str = "") -> int:
    """Deterministic seed from the run context, so a re-run reproduces the draw.

    Never seed from the clock, and never share one seed across stages: two stages
    drawing from the same seeded generator become order-dependent on each other.
    """
    h = hashlib.sha256(f"{run_id}|{stage}|{extra}".encode()).digest()
    seed = int.from_bytes(h[:8], "big") % (2 ** 32 - 1)
    log.info("seed for stage=%s run=%s -> %d", stage, run_id, seed)
    return seed


def call_once(keys: pd.Series, ledger_path: str, call_fn) -> pd.DataFrame:
    """Call an external service only for keys not already in the ledger.

    The ledger is the record of side effects. A restart consults it and skips
    anything already done, which is what makes a paid or rate-limited call safe
    to re-run.
    """
    if os.path.exists(ledger_path):
        ledger = pd.read_parquet(ledger_path)
    else:
        ledger = pd.DataFrame({"key": pd.Series(dtype="string")})
    todo = pd.Index(keys.dropna().unique()).difference(pd.Index(ledger["key"]))
    log.info("external calls: %d keys, %d already in ledger, %d to call",
             len(keys.unique()), len(ledger), len(todo))
    if len(todo) == 0:
        return ledger
    fresh = call_fn(list(todo))
    out = pd.concat([ledger, fresh], ignore_index=True).drop_duplicates("key")
    tmp = ledger_path + ".tmp"
    out.to_parquet(tmp, index=False)
    os.replace(tmp, ledger_path)
    return out


def stable_sjoin_nearest(left: gpd.GeoDataFrame, right: gpd.GeoDataFrame,
                         left_key: str, right_key: str) -> gpd.GeoDataFrame:
    """Nearest join with deterministic tie-breaking.

    Equidistant candidates otherwise resolve by whatever order the spatial index
    yielded, which can differ between library versions and between runs.
    """
    left = left.sort_values(left_key).reset_index(drop=True)
    right = right.sort_values(right_key).reset_index(drop=True)
    joined = gpd.sjoin_nearest(left, right, how="left", distance_col="_d")
    joined = (joined.sort_values([left_key, "_d", right_key])
                    .groupby(left_key, as_index=False).first())
    return joined


def assert_idempotent(stage_fn, ctx, out_path: str) -> None:
    """Run a stage twice into separate paths and assert the outputs match."""
    a = stage_fn(ctx)
    first = pd.read_parquet(a)
    os.remove(a)
    b = stage_fn(ctx)
    second = pd.read_parquet(b)
    if not first.equals(second):
        raise AssertionError(f"stage output differs between runs: {out_path}")
    log.info("idempotency asserted for %s (%d rows)", out_path, len(first))

The assert_idempotent helper is worth running in continuous integration on every stage. Idempotency is a property people believe they have and frequently do not, and the only reliable way to know is to execute the stage twice and compare. It is a cheap test that catches the appended row, the unseeded draw and the unstable tie in one go.

Deriving the seed from the run identifier and the stage name, rather than from a single pipeline-level seed, keeps stages independent. With one shared generator, adding a stage that consumes random numbers changes the draws every later stage receives, and the pipeline’s output shifts for a reason that has nothing to do with the data.

One Seed Couples Every Stage to Every Other Two pipeline layouts. In the first, a single seeded generator is passed through three stages in sequence, so inserting a new stage between the first and second changes every draw the later stages receive and their outputs shift for no data-related reason. In the second, each stage derives its own seed from the run identifier and its own name, so stages are independent and adding one leaves the others' draws untouched. A note records that the second layout also makes a single stage reproducible in isolation, which the first does not. Per-stage seeds keep a change local Shared generator mask points add noise bootstrap insert a stage here and all three change Derived per stage mask points add noise bootstrap each seeded from run id plus its name Independent seeds also let one stage be reproduced alone, which the shared generator forbids

Validation & Edge Cases

Incremental stages that must append. Some outputs genuinely accumulate, such as a geocode cache. Make the accumulation itself idempotent by writing with a deduplicating key and treating the cache as an input whose version is recorded, so a re-run with the same inputs produces the same cache state.

Floating-point non-determinism. Parallel reductions can sum in different orders and produce results differing in the last bits. For most surveillance output this is irrelevant; for an equality assertion it fails the test. Compare with a tolerance, or force a deterministic reduction order where exact equality matters.

Library-version-dependent geometry. A projection library upgrade can shift coordinates slightly, so a stage that was idempotent across runs is not idempotent across an upgrade. Pin versions and treat an upgrade as a deliberate change with a regression comparison.

Partial external calls. A batch geocode that fails halfway has charged for the completed portion. Writing the ledger incrementally, rather than once at the end, means a restart resumes rather than repeats.

Restart after a partial partition replace. The staging-then-swap pattern leaves a .staging directory if the process dies mid-write. The next run must remove it rather than reusing it, or a half-built partition becomes the output.

INFO pipeline.idempotent seed for stage=geomask run=20260918-4f2a1c8e -> 2381044917
INFO pipeline.idempotent external calls: 1102 keys, 1006 already in ledger, 96 to call
INFO pipeline.idempotent partition replaced: /data/runs/as_of=2026-09-17/geocoded rows=1184
INFO pipeline.idempotent idempotency asserted for geocoded (1184 rows)

The staging-then-swap pattern is what makes the partition replace safe to interrupt, and its states are worth laying out:

Staging Then Swap, and What an Interruption Leaves Three states of a partition replace. First, the new data is written to a staging directory beside the live partition, while readers continue to see the old partition. Second, the old partition is removed and the staging directory is renamed into its place in one step. Third, readers see the new partition. If the process dies during the first state, the live partition is untouched and a stale staging directory remains, which the next run must delete rather than reuse. There is no state in which readers see a half-written partition. Readers never see a half-written partition 1 · Write staging live partition unchanged new data beside it readers: old data 2 · Swap remove old, rename staging into place a single step 3 · Done partition fully replaced no rows appended readers: new data killed here: stale staging next run deletes, never reuses every interruption point leaves either the old or the new partition intact

Compliance Notes

Record the derived seed for every stochastic stage alongside the run identifier, because a disclosure control step’s reproducibility depends on it — a reviewer regenerating a geomasked release needs the exact seed the run used, and a seed derived from the run identifier and stage name is reconstructable from the provenance row alone.

The external-call ledger deserves attention as a disclosure surface: a geocode ledger keyed by normalised address is a table of addresses, which belongs inside the protected environment under the case file’s controls and retention schedule. Where a ledger must be shared between environments — a cache reused across projects, for instance — it should be keyed by a salted hash of the address rather than the address itself, with the salt held with the same care as any other key.