13 min read2,926 words

Building an Incremental Ingestion Pipeline for GIS Updates

A utility asset register is never finished being edited. Field crews close out work orders, surveyors correct hydrant positions, a CAD conversion drops a new subdivision into the system, and a SCADA integration re-stamps device attributes overnight. When the pipeline that carries those edits into the enterprise geodatabase is a nightly full reload, every one of those small changes forces the whole dataset to be truncated and rewritten — hours of processing, a full-table lock that blocks editors, a topology rebuild across features that never changed, and a blast radius where a single malformed source row corrupts assets that were perfectly fine yesterday. At the scale of a real network the full reload stops being merely slow and becomes unsafe: it is impossible to say which assets actually changed, so it is impossible to audit what an overnight run did. The discipline that fixes this belongs to the data ingestion pipelines for utility assets practice within core utility GIS fundamentals: move from reloading everything to ingesting only what changed.

Incremental ingestion — change-data-capture, in database terms — reframes each run as a small, auditable set of deltas applied idempotently on top of a known prior state. This page builds that pipeline end to end: a delta detector that classifies every incoming record by content hash and timestamp, an idempotent upsert that touches only the rows that moved, and a topology validation pass scoped to the changed extent so the network is proven consistent after each batch rather than hoped to be. The result is a run that completes in seconds instead of hours, names exactly what it changed, and refuses to advance if it broke the graph.

Environment Prerequisites

Lock these before the first incremental run; the delta model depends on a stable key and a trustworthy prior state, and both must exist before any comparison is meaningful.

  • Software: ArcGIS Pro 3.2+ (Standard or Advanced) for arcpy.un.ValidateTopology and versioned editing, with Python 3.11 and pandas>=2.0, geopandas>=1.0, and shapely>=2.0 in an isolated conda environment. Install with conda install -c conda-forge geopandas pandas shapely.
  • A stable, source-assigned asset identifier. Change-data-capture is only correct when each asset carries a durable key that survives edits — a GlobalID or an authoritative facility ID, never the database ObjectID, which is reassigned on reload. Without a stable key an update is indistinguishable from a delete-plus-insert, and the topology churns needlessly.
  • A persisted run manifest. The pipeline needs somewhere to record, per asset, the content hash and last-modified timestamp of the last successful run. A single-table manifest in the geodatabase or a version-controlled sidecar file is sufficient; it is the watermark the next run compares against.
  • A single authoritative CRS for the target. Every incoming feed is reprojected to the network’s frame before hashing geometry, following the CRS alignment and geodetic transformations discipline. Hashing coordinates in mixed projections makes identical assets look changed and defeats delta detection entirely.
  • A versioned, isolated workspace. Apply deltas to a named version or a snapshot geodatabase, never the default version under an active edit session, so a batch can be reconciled and posted as a unit — or rolled back cleanly if validation fails.
  • A topology that is currently validated. Start from a clean slate with no dirty-area backlog, so the only errors the post-batch validation surfaces are the ones this batch introduced.

Schema-Aware Delta Protocol — Run Before the Upsert

Most incremental-ingestion failures are not in the write; they are in a delta that was computed wrong, so the pipeline confidently applies the wrong changes. Work this ordered checklist before any upsert touches the geodatabase; the earliest item is the most frequent culprit.

  1. Confirm the stable key is present and unique on the incoming batch. A null or duplicated asset id makes every downstream comparison ambiguous. Reject the batch if the key column has nulls or duplicates rather than guessing which record wins — a bad key corrupts the manifest for every future run.
  2. Normalize before hashing. Trailing whitespace, inconsistent numeric precision, and coordinate order all change a content hash without changing meaning, producing phantom updates. Round coordinates to the network tolerance, canonicalize attribute types, and sort geometry vertices deterministically before computing the hash so only real edits register.
  3. Reproject to the target CRS first. Compute the geometry hash in the authoritative frame, not the source frame. Two identical hydrants delivered in different projections must hash the same; if they do not, delta detection reports the entire feed as changed.
  4. Classify deletes explicitly, and prefer tombstones. An asset absent from the incoming batch is ambiguous — it may be retired, or the source may simply be a partial extract. Treat absence as a delete only when the feed is authoritative and complete; otherwise require an explicit retirement flag so a partial extract never silently deletes live assets.
  5. Bound the batch size. A delta that suddenly spans a large fraction of the network is a red flag for an upstream schema change or a CRS regression, not a real edit burst. Cap the batch and halt for review when the change ratio exceeds a threshold, so a broken source cannot rewrite the whole register unattended.

Minimal Reproducible Implementation

The following is the complete change-data-capture core: it loads the prior manifest, classifies every incoming record as inserted, updated, unchanged, or deleted by comparing a normalized content hash and timestamp, applies only the changes as an idempotent upsert keyed on the stable asset id, and returns a structured report. Re-running the same batch is a no-op — every write is keyed and content-checked, so idempotency is a property of the algorithm rather than a hope.

import hashlib
import json
import logging
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone

import geopandas as gpd
import pandas as pd

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")


@dataclass
class IngestReport:
    """Structured, gateable outcome of one incremental batch."""
    inserted: list[str] = field(default_factory=list)
    updated: list[str] = field(default_factory=list)
    unchanged: int = 0
    deleted: list[str] = field(default_factory=list)
    change_ratio: float = 0.0
    ok: bool = True
    message: str = ""


def _content_hash(row: pd.Series, attr_cols: list[str], tol: float = 0.001) -> str:
    """Normalize attributes + geometry to a deterministic hash.

    Coordinates are rounded to the network tolerance and attributes are cast to
    a canonical string form so cosmetic differences do not register as edits.
    """
    attrs = "|".join(f"{c}={row[c]!r}" for c in sorted(attr_cols))
    geom = row.geometry
    coords = () if geom is None or geom.is_empty else tuple(
        (round(x, 3), round(y, 3)) for x, y in geom.exterior.coords
    ) if geom.geom_type == "Polygon" else tuple(
        (round(x, 3), round(y, 3)) for x, y in geom.coords
    )
    payload = f"{attrs}#{coords}".encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def detect_and_upsert(
    incoming: gpd.GeoDataFrame,
    manifest: dict[str, dict],
    key: str,
    attr_cols: list[str],
    authoritative_complete: bool,
    max_change_ratio: float = 0.40,
) -> tuple[gpd.GeoDataFrame, dict[str, dict], IngestReport]:
    """Classify deltas and apply an idempotent upsert keyed on `key`.

    Returns the rows to write, the advanced manifest, and a structured report.
    Only inserted/updated rows are returned for writing; deletes are reported
    as tombstones. The manifest advances only when the caller commits.
    """
    report = IngestReport()

    if incoming[key].isna().any() or incoming[key].duplicated().any():
        report.ok = False
        report.message = f"batch has null or duplicate {key} values"
        return incoming.iloc[0:0], manifest, report

    now = datetime.now(timezone.utc).isoformat()
    incoming = incoming.copy()
    incoming["_hash"] = incoming.apply(lambda r: _content_hash(r, attr_cols), axis=1)

    to_write_ids: list[str] = []
    new_manifest = dict(manifest)
    incoming_ids = set()

    for _, row in incoming.iterrows():
        aid = str(row[key])
        incoming_ids.add(aid)
        prior = manifest.get(aid)
        if prior is None:
            report.inserted.append(aid)
            to_write_ids.append(aid)
        elif prior["hash"] != row["_hash"]:
            report.updated.append(aid)
            to_write_ids.append(aid)
        else:
            report.unchanged += 1
        new_manifest[aid] = {"hash": row["_hash"], "seen": now}

    # Deletes: assets in the manifest but absent from an authoritative full feed.
    if authoritative_complete:
        for aid in list(manifest.keys()):
            if aid not in incoming_ids:
                report.deleted.append(aid)
                new_manifest.pop(aid, None)

    changed = len(report.inserted) + len(report.updated) + len(report.deleted)
    report.change_ratio = changed / max(len(manifest), 1)
    if report.change_ratio > max_change_ratio:
        report.ok = False
        report.message = (
            f"change ratio {report.change_ratio:.0%} exceeds {max_change_ratio:.0%}; "
            "halting for review (possible schema or CRS regression)"
        )
        return incoming.iloc[0:0], manifest, report  # keep old manifest

    to_write = incoming[incoming[key].astype(str).isin(to_write_ids)].drop(columns=["_hash"])
    logging.info(
        "delta: +%d ~%d =%d -%d (ratio %.1f%%)",
        len(report.inserted), len(report.updated), report.unchanged,
        len(report.deleted), report.change_ratio * 100,
    )
    return to_write, new_manifest, report


if __name__ == "__main__":
    from shapely.geometry import Point

    prior = {"HYD-001": {"hash": "stale", "seen": "2026-07-12T00:00:00+00:00"}}
    batch = gpd.GeoDataFrame(
        {"facility_id": ["HYD-001", "HYD-002"], "flow_gpm": [1500, 1200]},
        geometry=[Point(500010.0, 3100020.0), Point(500110.0, 3100120.0)],
        crs="EPSG:26917",
    )
    rows, manifest, report = detect_and_upsert(
        batch, prior, key="facility_id",
        attr_cols=["flow_gpm"], authoritative_complete=True,
    )
    print(json.dumps(asdict(report), indent=2))

Only the rows in to_write reach the geodatabase, and each write is an upsert keyed on facility_id, so applying the same batch twice changes nothing the second time. The manifest is returned but not yet persisted: the pipeline advances the watermark only after the batch is validated, which is the next and non-negotiable step. The following diagram shows the loop those three stages form.

The incremental ingestion loop from delta detection to validated commit A left-to-right flow. An incoming source batch and the prior-run manifest both enter a delta detector that classifies records as inserted, updated, unchanged, or deleted. Only changed rows pass to an idempotent upsert into a versioned workspace, which feeds a topology validation gate scoped to the changed extent. If validation passes, the manifest advances and a run report is written; if it fails, the batch is rolled back and the manifest is left unchanged. Source batch incoming feed Prior manifest hash + timestamp Delta detect +ins ~upd =same -del Idempotent upsert versioned workspace Validate topology changed extent Advance manifest Roll back no advance pass fail

Applying deltas without re-proving the network is how incremental ingestion becomes faster than full reloads and less trustworthy — a moved valve or a new lateral can introduce a disconnected edge that the full-reload topology rebuild would have caught. The batch must therefore end in a validation pass scoped to the changed extent, and the manifest must advance only if it succeeds.

import arcpy
import logging


def validate_batch_topology(topology: str, changed_extent) -> dict:
    """Validate topology over the changed extent and report new errors.

    Returns a structured result; `ok` is False when the batch introduced
    dirty areas or connectivity errors, which must block the manifest advance.
    """
    try:
        arcpy.management.ValidateTopology(topology, extent=changed_extent)
        errors = arcpy.management.ExportTopologyErrors(topology, "memory", "batch_topo_err")
        new_errors = 0
        with arcpy.da.SearchCursor(errors.getOutput(0), ["isException"]) as cur:
            for (is_exception,) in cur:
                if not is_exception:
                    new_errors += 1
        ok = new_errors == 0
        logging.info("post-batch validation: %d unresolved topology errors", new_errors)
        return {"ok": ok, "topology_errors": new_errors,
                "action": "advance manifest" if ok else "roll back batch"}
    except arcpy.ExecuteError as exc:
        logging.error("topology validation failed: %s", exc)
        return {"ok": False, "topology_errors": -1, "action": "roll back batch"}

Scoping validation to changed_extent is what keeps the incremental win: only the assets this batch touched, plus their immediate neighbors, are re-checked, so validation cost tracks the size of the delta rather than the size of the network. The same connectivity guarantees are enforced end to end by the topology and tracing workflows that consume the register — an incremental pipeline that skips this step simply defers the corruption to the first trace that crosses the broken edge.

Production Deployment Pattern

A delta detector in a notebook is a prototype; an enforced, restartable pipeline is an engineering control. Promote it as follows:

  1. Persist the manifest transactionally with the commit. Advance the watermark in the same reconcile/post transaction that writes the features, so a crash between the write and the manifest update cannot leave the two disagreeing. A run that fails validation leaves both the data and the manifest untouched and is safely retried.
  2. Wire the report into CI/CD. Run detect_and_upsert followed by validate_batch_topology from GitHub Actions or Jenkins on every source drop and on a schedule. Let a False ok fail the build and open a remediation ticket; parse the IngestReport to post the exact inserted, updated, and deleted asset ids so an operator sees what a batch did without opening the map.
  3. Make source reads retry with backoff. Enterprise geodatabase and REST feeds fail intermittently under load. Wrap each read in a bounded exponential-backoff retry so a transient lock produces a retry rather than a spurious empty batch that the delta detector would read as a mass delete.
  4. Push validated changes onward by delta. Once the batch passes, propagate only the changed asset ids to the CMMS over its REST endpoint and to any downstream trace cache, so the incremental discipline carries through the whole chain instead of triggering a full downstream rebuild. This is the same watermark-driven pattern that lets an incremental topology rebuild after field edits touch only what moved.
  5. Keep an immutable run log. Append each run’s report — counts, change ratio, topology-error count, source CRS, and library versions — to a timestamped, version-controlled log. That chain of custody is what lets a reliability or rate-case review reconstruct exactly which assets changed on any given night and prove the network was validated before the batch was accepted.

Conclusion

Incremental ingestion replaces the blunt nightly full reload with a small, auditable, idempotent set of deltas. Detecting change by normalized content hash and timestamp keeps phantom updates out; keying every write on a stable asset id makes re-running a batch a no-op; scoping topology validation to the changed extent keeps the run fast while still proving the network consistent; and advancing the manifest only after validation passes makes the whole pipeline restartable and safe. Built this way, an overnight run finishes in seconds, names precisely what it changed, and refuses to corrupt the graph — turning data ingestion from a risky maintenance window into a routine, auditable control.

For authoritative reference, consult the ArcGIS Pro geodatabase topology documentation and the pandas documentation.