Spatial Epi Public Health GIS

Orchestrating Spatial Surveillance Pipelines

This guide is part of the Surveillance Pipeline Automation & Spatial Data Quality section. It covers the control flow around the analysis: what runs when, what happens when a stage fails, and what property each stage must have for the whole thing to be operable by someone who did not write it.

Concept & Epidemiological Alignment

Orchestration is the discipline of arranging computations so that partial failure is survivable. In surveillance that matters more than in most domains, because the pipeline runs unattended against data that arrives on somebody else’s schedule and because its output feeds decisions with a clock on them.

The organising concept is the idempotent stage: a unit of work that produces the same output when run twice on the same inputs and leaves no state behind that changes the second run’s behaviour. Idempotency is what makes restart safe, and restart is what makes a nightly pipeline operable — an operator who finds a failed run at 6 a.m. needs to be able to re-run it without reasoning about what the first attempt half-completed.

The second concept is the explicit dependency graph. Stages declare what they consume and produce, and the orchestrator derives the order. That derivation is what lets a failure in geocoding skip the analysis and publication stages rather than running them on yesterday’s geocodes, which is the failure mode that produces confidently wrong surveillance output.

The third is run identity. Every execution carries an identifier that appears in the provenance record, in every artifact, and in every alert. A surveillance system without run identity can answer “what does the map say” and cannot answer “why”, and the second question is the one that arrives with a phone call.

None of this is specific to a particular orchestrator. The properties matter; the tool that provides them is a local choice, and a well-structured set of stages can be driven by a workflow engine, a make-style dependency tool or a shell script with the same guarantees.

Stage Design Patterns

Pattern What it gives When it is wrong
Pure stage: read inputs, write owned outputs Re-runnable, testable, restartable Never wrong; the cost is intermediate storage
In-place update of a shared table Saves storage Destroys re-runnability and history
Atomic write: temp then move No truncated outputs Requires same-filesystem or atomic object copy
Partition per run date Backfills cannot touch other dates Many small files if the date grain is too fine
Fan-out by geography Parallelism, partial success Cross-boundary analyses break if fanned naively
Single monolithic job Simple to schedule One failure costs the entire run
A Failure Should Skip, Not Poison, Downstream Work A dependency graph of pipeline stages. Ingestion succeeds and feeds two branches. The geocoding branch fails, so the cluster analysis and publication stages that depend on it are skipped rather than run on stale geocodes. A second branch computing healthcare access metrics depends only on facility and network data, so it continues and publishes normally. The diagram shows that the value of an explicit dependency graph is exactly this: partial output that is correct, rather than complete output that is silently wrong. Correct partial output beats complete output built on stale inputs Ingest succeeded Geocode FAILED Cluster scan skipped Publish map skipped Access metrics independent branch Suppress ran normally Publish tables succeeded Without the graph, the cluster scan would have run against the previous night's geocodes

Spatial Data Prerequisites

  • Stage inputs declared explicitly, including the vintage of every reference layer, so the orchestrator can detect a changed input and so a re-run reads the same data.
  • A working coordinate reference system fixed for the pipeline and asserted at every stage boundary.
  • A storage layout partitioned by run date, so a backfill writes into its own partition and cannot disturb the current one.
  • A provenance table with one row per run: identifier, wall-clock window, input vintages, code commit, library versions, parameters, seeds and gate outcomes.
  • Atomic write capability in the target store — a temporary path and a rename, or an object-store copy that is atomic at the key level.
  • A defined time zone for every window computation, stored with the run rather than inherited from the host.

Production Implementation

The skeleton below implements the properties independently of any orchestrator: stages declare inputs and outputs, results are written atomically into a run-scoped partition, and a provenance row is emitted whether the run succeeds or fails.

# Orchestration primitives: idempotent stages, atomic writes, run provenance.
# Pinned: pandas==2.2.2, geopandas==0.14.4
import hashlib
import json
import logging
import os
import shutil
import subprocess
import tempfile
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone

import pandas as pd

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


@dataclass
class RunContext:
    run_id: str
    as_of: pd.Timestamp                 # the logical date this run represents
    root: str                           # partition root, e.g. /data/runs
    inputs: dict = field(default_factory=dict)   # name -> {"path":…, "vintage":…}
    params: dict = field(default_factory=dict)
    seeds: dict = field(default_factory=dict)

    @property
    def partition(self) -> str:
        return os.path.join(self.root, f"as_of={self.as_of.date()}", self.run_id)

    def config_digest(self) -> str:
        payload = {"inputs": self.inputs, "params": self.params, "seeds": self.seeds,
                   "as_of": str(self.as_of.date())}
        return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()


def new_run(as_of: pd.Timestamp, root: str, **kw) -> RunContext:
    rid = f"{as_of.date():%Y%m%d}-{uuid.uuid4().hex[:8]}"
    ctx = RunContext(run_id=rid, as_of=as_of, root=root, **kw)
    os.makedirs(ctx.partition, exist_ok=True)
    log.info("run %s started, as_of=%s partition=%s", rid, as_of.date(), ctx.partition)
    return ctx


def atomic_write(df: pd.DataFrame, path: str) -> None:
    """Write to a temporary file in the same directory, then move into place.

    A run killed mid-write must never leave a truncated file that the next stage
    will happily read as complete.
    """
    d = os.path.dirname(path)
    os.makedirs(d, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=d, suffix=".tmp")
    os.close(fd)
    try:
        df.to_parquet(tmp, index=False)
        os.replace(tmp, path)           # atomic on the same filesystem
    finally:
        if os.path.exists(tmp):
            os.remove(tmp)


def stage(name: str, produces: str):
    """Decorator: skip if the output exists, write atomically, log timing."""
    def wrap(fn):
        def inner(ctx: RunContext, *a, **kw):
            out = os.path.join(ctx.partition, produces)
            if os.path.exists(out):
                # Idempotency: a completed stage is not redone on restart.
                log.info("stage %-16s SKIP (output present) run=%s", name, ctx.run_id)
                return out
            t0 = datetime.now(timezone.utc)
            log.info("stage %-16s START run=%s", name, ctx.run_id)
            result = fn(ctx, *a, **kw)
            atomic_write(result, out)
            dt = (datetime.now(timezone.utc) - t0).total_seconds()
            log.info("stage %-16s OK rows=%d seconds=%.1f run=%s",
                     name, len(result), dt, ctx.run_id)
            return out
        inner.stage_name = name
        return inner
    return wrap


def write_provenance(ctx: RunContext, gates: dict, status: str,
                     libs: dict | None = None) -> None:
    """One row per run — emitted on success AND on failure."""
    row = {
        "run_id": ctx.run_id,
        "as_of": str(ctx.as_of.date()),
        "status": status,
        "started_utc": datetime.now(timezone.utc).isoformat(),
        "code_commit": _git_commit(),
        "libraries": json.dumps(libs or {}, sort_keys=True),
        "inputs": json.dumps(ctx.inputs, sort_keys=True),
        "params": json.dumps(ctx.params, sort_keys=True),
        "seeds": json.dumps(ctx.seeds, sort_keys=True),
        "gates": json.dumps(gates, sort_keys=True),
        "config_sha256": ctx.config_digest(),
    }
    atomic_write(pd.DataFrame([row]),
                 os.path.join(ctx.partition, "_provenance.parquet"))
    log.info("run %s %s config_sha256=%s", ctx.run_id, status, row["config_sha256"][:16])


def _git_commit() -> str:
    try:
        return subprocess.check_output(["git", "rev-parse", "HEAD"],
                                       text=True, stderr=subprocess.DEVNULL).strip()
    except Exception:
        return "unknown"

Two properties of that skeleton carry most of the value. The stage decorator skips work whose output already exists, which turns restart from a reasoning problem into a mechanical one — re-running a partially failed pipeline resumes exactly where it stopped. And provenance is written on failure as well as on success, because a failed run is precisely the one someone will ask about, and a record that exists only for successes cannot answer.

The run-scoped partition path is what makes backfills safe. A backfill for a date three months ago writes into that date’s partition under its own run identifier and cannot touch anything the nightly run produces, so the two can execute concurrently without a lock.

Parameter Selection & Tuning

Stage granularity. Too coarse and a failure costs an hour of recomputation; too fine and the orchestration overhead and intermediate storage dominate. A useful rule is that a stage should be the largest unit of work you would be willing to repeat in full.

Schedule offset. Scheduling at the moment upstream data is expected guarantees a race. Offset the run by enough that late-arriving feeds have landed, and assert the input’s own timestamp rather than trusting the schedule.

Retry policy. Retry transient failures — network, rate limits, lock contention — with backoff, and never retry an assertion failure, which will fail identically every time and only delays the alert.

Concurrency limits. A geocoding stage that fans out can exhaust an external service’s rate limit and get the agency’s key throttled. Cap concurrency explicitly rather than relying on the orchestrator’s defaults.

Backfill window and order. Backfills should run oldest-first when later dates depend on earlier ones, and in any order when they do not. Declare which regime applies; the wrong assumption silently produces inconsistent history.

Edge Cases & Failure Modes

A stage that is not actually idempotent. Appending to a shared table, incrementing a counter, or calling an external service that charges per call are all operations whose second execution differs from the first. Either make them idempotent with a deduplication key, or isolate them behind a stage that records completion.

Inputs that change between stages. If the ingest stage reads a boundary file and the analysis stage reads it again, a mid-run update produces an inconsistent result. Snapshot inputs into the run partition, or record their content hash and assert it.

A backfill that overwrites corrections. Re-running an old date regenerates output from current code and current reference data, which can differ from what was published. That is usually the intent, but not always — a published figure that a partner agency has cited should not silently change. Version published artifacts and keep the original.

Clock skew across hosts. A distributed pipeline whose stages disagree about the current time will produce windows with gaps or overlaps. Derive every window from the run’s logical date, never from a stage’s own clock.

Long-running stages and orchestrator timeouts. A state-wide routing job that exceeds a default timeout is killed mid-write. Atomic writes contain the damage; realistic timeouts prevent it.

Alerting on every failure. A pipeline that pages on transient failures gets muted within a month. Alert on exhausted retries and on assertion failures, and record the rest.

What Idempotency Buys at Six in the Morning Two rows showing the same five-stage pipeline. The first row is the original run, in which the first three stages completed and the fourth failed, leaving the fifth unrun. The second row is the restart, in which the first three stages are skipped because their outputs already exist in the run partition, the fourth stage runs and succeeds, and the fifth then runs normally. The elapsed time of the restart is a fraction of the original, which is the practical reason idempotency matters to an operator rather than to a designer. Restart resumes; it does not recompute original run ingest geocode analyse suppress FAIL publish restart skip skip skip suppress publish Skipping is mechanical: the stage output already exists in the run partition which means the operator does not have to reason about what the first attempt half-finished and the same mechanism makes an accidental double-trigger harmless

The time-zone rule in that list prevents a failure that occurs predictably twice a year, and it is easiest to see on a clock:

Whose Midnight Defines Yesterday Two twenty-four hour bands for the same calendar date, one in local time and one in coordinated universal time, offset from each other by several hours. Records stamped in the offset hours fall into different days depending on which definition is used. On the two nights each year when daylight saving changes, the local day is twenty-three or twenty-five hours long and the disagreement changes size. Defining every window in one explicit time zone stored with the run removes the ambiguity. Define every window in one explicit time zone, stored with the run local 17 September, local UTC 17 September, UTC local only UTC only records in the shaded hours land on different days under each definition and the offset changes size on the two daylight-saving nights each year derive windows from the run's logical date, never from a host clock

Compliance & Audit Controls

The provenance row is the audit control. It must be written for every run regardless of outcome, be immutable once written, and contain enough to rebuild the run: input identifiers with vintages, code commit, library versions, all parameters, all seeds, the gate outcomes and the configuration digest.

Alerts are the other control, and they are a disclosure surface. A stack trace from a geocoding stage contains addresses; an assertion message from a suppression stage contains small-cell counts. Alert payloads must be restricted to run identifier, stage name, failure class and counts, with anything diagnostic left inside the protected environment and retrieved by someone with access.

Separate the three privilege levels. Reading published output, restarting a failed run and deploying new pipeline code are different authorities, and an orchestrator that runs everything as one identity makes the automation the broadest-privileged component in the system. Where the orchestrator needs credentials for an external geocoder or routing service, those belong in a secret store with rotation, not in the pipeline configuration that is committed alongside the code.

Production Implementation Checklist

Frequently Asked Questions

What is the smallest useful orchestration setup? A make-style dependency tool over a set of stages that write into a dated partition, plus a provenance file. That gives skip-on-completion, correct ordering and a record, which is most of the value. A full workflow engine adds scheduling, retries, observability and fan-out, which matter once the pipeline has more than a handful of stages or more than one operator.

How do I make a geocoding stage idempotent when it calls a paid API? Cache by normalised address. The stage looks up each address in a result table keyed by the normalised string and the geocoder vintage, calls the service only for misses, and writes results back. A re-run then costs nothing and returns identical coordinates, which is both cheaper and more reproducible.

Should a backfill use today’s code or the code from that date? Today’s, in nearly all cases — the point of a backfill is usually to apply a correction or an improved method to history. The exception is reproducing a specific published figure, which requires checking out that run’s commit. The provenance row is what makes that possible.

How do I test orchestration logic? With stages that do trivial work. A test pipeline with the same graph shape and stages that write a fixed row each exercises the ordering, skipping, atomicity and provenance without needing real data or real runtime, and it catches the orchestration bugs that are otherwise found in production at 6 a.m.

What should happen when a gate fails at the suppression stage? Nothing publishes. Suppression is a fail-closed gate, so an error there must stop the run and alert, never fall through to publication. The downstream stages skip, the provenance row records the failure, and yesterday’s published product remains the current one until someone resolves it.