This guide is part of Orchestrating Spatial ETL Pipelines, the reference for turning extract, clean, and load scripts into scheduled, restartable, observable spatial data products.

Scheduling and Incremental Spatial Loads

A spatial pipeline that re-downloads and re-cleans its entire source archive on every run is affordable exactly once — the first time, when the archive is small. By the time a daily satellite feed has a two-year history, or a national parcel portal holds forty million polygons, the full-reload strategy costs hours of compute and terabytes of transfer to discover that almost nothing changed. The job of the scheduler is not simply to fire the pipeline on a clock; it is to answer what is new since last time cheaply, process only that delta, and record how far it got so the next run can pick up exactly where this one stopped.

Two problems make this harder for spatial data than for ordinary tabular ETL. First, the source’s update rhythm rarely matches any clean cron cadence: a Sentinel-2 tile is revisited every five days but published hours later, a government portal refreshes “sometime Tuesday,” and an Overpass extract reflects edits made continuously. Second, spatial records arrive late and out of order — a scene processed upstream yesterday can appear in the catalog with yesterday’s acquisition timestamp today, so a naive “give me everything after my last run” query silently skips it. This guide covers the machinery that solves both: watermark tracking, change detection, catch-up and backfill runs, event sensors, overlap windows for late data, and the idempotent upserts that make re-processing safe.

Schedule Cadence Versus Source Update Frequency

The first design decision is the relationship between how often the pipeline runs and how often the source actually changes. These are almost never the same number, and conflating them is the root cause of both missed data and wasted runs. The diagram below shows the watermark model that decouples them: the scheduler ticks on a fixed cadence, but each tick only processes the window of source updates newer than the stored watermark, and a detected gap is backfilled as a separate bounded run.

The rule that follows from this model: schedule at least as often as your tightest freshness requirement, and detect change on every tick. If stakeholders need data no more than six hours stale, run every hour or two — not every six — so a missed or failed run still leaves you inside the window. Because each tick is gated behind change detection, the extra runs that find nothing new cost only a cheap metadata check, not a reprocess. This inverts the intuition from batch tabular ETL, where you run exactly as often as the data lands; with a moving spatial archive you run more often than it lands and let the watermark absorb the difference.

Prerequisites & Environment

The patterns here are framework-agnostic — the watermark and change-detection logic lives in plain Python and a small state table — but the examples use the following stack. A relational store (PostgreSQL/PostGIS or SQLite for local runs) holds both the loaded spatial data and the watermark state table.

• Python 3.10+ — for modern type hints and datetime timezone handling.
• Core libraries — geopandas>=0.14, shapely>=2.0 (for the vectorized geometry API), sqlalchemy>=2.0, pystac-client>=0.7 for STAC-sourced discovery, and requests>=2.31 for HTTP conditional requests.
• A state store — any transactional database. The watermark table must be updated in the same transaction discipline as the load, or in a strictly-after step, never before.

pip install "geopandas>=0.14" "shapely>=2.0" "sqlalchemy>=2.0" \
    "pystac-client>=0.7" requests

Verify the timezone handling that incremental correctness depends on — watermarks must be stored and compared as timezone-aware UTC, never naive local time:

from datetime import datetime, timezone

now = datetime.now(timezone.utc)
assert now.tzinfo is not None, "watermarks must be timezone-aware UTC"

Version / Compatibility Matrix

Each orchestrator expresses the same watermark-and-delta model with different native primitives. You do not need any of them to implement incremental loads — the logic is portable — but if you already run one, use its idioms rather than fighting them.

Pin the geometry stack together: shapely>=2.0 with geopandas>=0.14 guarantees the vectorized shapely.is_valid(arr) / shapely.make_valid(arr) API used in the load-validation step below.

Step-by-Step Implementation

Step 1 — Persist and Read a Watermark

The watermark is a single durable value per source stream: the high-water mark of what has been successfully loaded. Store it in a transactional table keyed by a source identifier so one pipeline can track many independent streams (one per collection, per tile group, per portal).

from __future__ import annotations

from datetime import datetime, timezone

from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine


def read_watermark(engine: Engine, source_id: str, default: datetime) -> datetime:
    """Return the stored high-water mark for a source, or a floor default."""
    with engine.connect() as conn:
        row = conn.execute(
            text("SELECT watermark_ts FROM etl_watermark WHERE source_id = :sid"),
            {"sid": source_id},
        ).fetchone()
    if row is None or row[0] is None:
        return default
    ts: datetime = row[0]
    # Normalize to timezone-aware UTC — a naive value here corrupts every comparison
    return ts if ts.tzinfo else ts.replace(tzinfo=timezone.utc)

The default (a floor like datetime(2020, 1, 1, tzinfo=timezone.utc)) is what a first-ever run uses, and it doubles as a documented starting point for a full historical backfill.

Step 2 — Detect Change Before Downloading

Not every source exposes a clean timestamp to compare against. Match the detection method to what the source offers, cheapest first, so most runs short-circuit before transferring any payload.

import requests


def source_changed(url: str, stored_etag: str | None,
                   stored_modified: str | None) -> tuple[bool, dict[str, str]]:
    """Conditional GET: return (changed, new_validators) without downloading the body."""
    headers: dict[str, str] = {}
    if stored_etag:
        headers["If-None-Match"] = stored_etag
    if stored_modified:
        headers["If-Modified-Since"] = stored_modified

    resp = requests.head(url, headers=headers, timeout=30, allow_redirects=True)
    if resp.status_code == 304:
        return False, {}
    validators = {
        k: resp.headers[k]
        for k in ("ETag", "Last-Modified")
        if k in resp.headers
    }
    return True, validators

For file- and portal-based sources this HTTP validator check is the entire change-detection story; the mechanics of persisting and comparing those validators are covered in detecting dataset changes with ETag and Last-Modified. For catalog-based sources, change detection is instead a datetime filter on the search query, described in Step 3. For a database source you already control, SELECT max(updated_at) is the equivalent probe.

Step 3 — Query Only the Delta, With an Overlap Window

With a watermark in hand, scope the extract to records strictly newer than it — minus a deliberate overlap so late-arriving records are not skipped. For a STAC source the delta is a datetime-range search; the paginated traversal of that search is handled by syncing STAC catalogs with pystac-client.

from datetime import datetime, timedelta

from pystac_client import Client
from pystac import Item


def fetch_delta_items(
    stac_client: Client,
    collections: list[str],
    bbox: tuple[float, float, float, float],
    watermark: datetime,
    overlap: timedelta = timedelta(hours=48),
) -> list[Item]:
    """Fetch STAC items updated since (watermark - overlap) to capture late arrivals."""
    lower = watermark - overlap
    search = stac_client.search(
        collections=collections,
        bbox=list(bbox),
        # 'updated' reflects catalog publication, which lags acquisition — the
        # field that actually moves when a scene appears late.
        query={"updated": {"gte": lower.isoformat()}},
        limit=100,
    )
    return list(search.items())

The overlap window is the single most important defence against silent gaps. Size it to the source’s worst-case publication lag: 24–48 hours is typical for satellite catalogs, minutes for a database CDC feed. The overlap only works because the load in Step 5 is idempotent — re-reading a record already loaded must be harmless.

Step 4 — Transform and Validate the Delta

Run CRS normalization and geometry repair on the delta exactly as a full load would, using the vectorized Shapely 2.0 API rather than per-row .apply(). Validation splits the batch rather than raising, so one malformed feature never blocks the watermark from advancing.

import geopandas as gpd
import shapely


def validate_delta(gdf: gpd.GeoDataFrame) -> tuple[gpd.GeoDataFrame, gpd.GeoDataFrame]:
    """Split a delta batch into (loadable, dead_letter) using the array geometry API."""
    geom = gdf.geometry.values
    valid = shapely.is_valid(geom) & ~gpd.GeoSeries(geom).is_empty.values
    passed = gdf.loc[valid].copy()
    dead = gdf.loc[~valid].copy()
    return passed, dead

Step 5 — Upsert Idempotently, Then Advance the Watermark

The load and the watermark advance form a strict ordering: upsert the delta keyed on a stable identifier, and only if that succeeds write the new watermark. Reversing the order — advancing first — is how pipelines silently drop data after a mid-load crash.

def upsert_and_advance(
    engine: Engine,
    gdf: gpd.GeoDataFrame,
    table: str,
    source_id: str,
    new_watermark: datetime,
    key: str = "scene_id",
) -> int:
    """Delete-then-insert the batch and advance the watermark in one transaction."""
    ids = gdf[key].unique().tolist()
    with engine.begin() as conn:  # one transaction: load + watermark are atomic together
        conn.execute(
            text(f"DELETE FROM {table} WHERE {key} = ANY(:ids)"),
            {"ids": ids},
        )
        gdf.to_postgis(table, conn, if_exists="append", index=False)
        conn.execute(
            text(
                "INSERT INTO etl_watermark (source_id, watermark_ts) "
                "VALUES (:sid, :ts) "
                "ON CONFLICT (source_id) DO UPDATE SET watermark_ts = EXCLUDED.watermark_ts"
            ),
            {"sid": source_id, "ts": new_watermark},
        )
    return len(gdf)

Deriving new_watermark from the maximum source timestamp actually observed in the batch — not from datetime.now() — is what keeps the overlap window meaningful and prevents the watermark from racing ahead of data the source has not yet published. The focused, reusable form of this whole read-fetch-upsert-advance loop is the subject of incremental spatial loading with watermark timestamps.

Advanced Patterns & Edge Cases

Sensors: Triggering on New Items Instead of a Clock

A fixed schedule wastes runs when a source is quiet and adds latency when it bursts. An event sensor inverts control: it polls a lightweight signal — new STAC items in a window, a new object under an S3 prefix, a bumped ETag — and only then launches the heavy pipeline. Airflow expresses this with deferrable sensors that release the worker slot while waiting; Dagster with sensors that yield RunRequests keyed on newly discovered partitions. The sensor’s own cursor is itself a watermark — the last item ID or timestamp it has already fired for — so the same persistence discipline applies. Asset-partitioned sensors and their backfills are covered in managing spatial assets with Dagster.

Catch-Up and Backfill Runs

Catch-up handles the recent past — the intervals a paused or failed scheduler missed — by replaying each missed window against its own watermark slice. Backfill handles the distant past — reprocessing history after a logic change or to seed a new target. Both are only cheap when the pipeline is partitioned so a run for interval N touches only interval N’s data. In Airflow this is catchup=True with interval-scoped queries; the deeper idempotency guarantees that make a re-run safe are detailed in writing idempotent spatial ETL tasks in Airflow.

Late-Arriving and Out-of-Order Data

Beyond the fixed overlap window, some sources emit records timestamped days or weeks in the past — a re-processed historical scene, a corrected parcel record. A pure overlap cannot catch these without becoming a full scan. The pragmatic pattern is a two-track schedule: a frequent incremental run with a short overlap for freshness, plus an infrequent (weekly) reconciliation run that widens the window dramatically or compares source-side max(updated_at) per partition against loaded counts to detect drift. Both tracks share the same idempotent upsert, so they can overlap in time without corrupting each other.

Performance Optimization

The payoff of incremental loading is measured against the full-reload baseline. The comparison below runs the same source through both strategies and reports the ratio that matters — bytes and rows actually processed per run once the archive is mature.

import time


def compare_load_cost(engine: Engine, run_full, run_incremental) -> dict[str, float]:
    """Benchmark a full reload against an incremental delta load on the same source."""
    t0 = time.perf_counter()
    full_rows = run_full()          # re-reads and rewrites the entire archive
    full_s = time.perf_counter() - t0

    t1 = time.perf_counter()
    delta_rows = run_incremental()  # reads only records newer than the watermark
    delta_s = time.perf_counter() - t1

    return {
        "full_rows": full_rows,
        "delta_rows": delta_rows,
        "full_seconds": round(full_s, 2),
        "delta_seconds": round(delta_s, 2),
        # On a mature archive this speedup grows without bound as history accumulates
        "speedup": round(full_s / delta_s, 1) if delta_s else float("inf"),
    }

The result is not a fixed percentage — it grows with history. On a two-year daily feed, an incremental run touches one day’s worth of the archive while a full reload touches all 730; the speedup is proportional to the accumulated history, which is exactly why the strategy becomes more valuable over time, not less. Keep the delta query itself indexed on the source timestamp so the “give me rows after the watermark” scan is a range seek, not a full-table scan that erases the saving.

Integration into ETL Pipelines

Wire the watermark table into the same object-storage and target layout the rest of the pipeline uses, and treat it as first-class state: back it up with the target, and expose its current value as a metric so an operator can see per-source freshness at a glance. Route delta records that fail validation to the dead-letter store rather than blocking the watermark advance — a poisoned record must never freeze a stream. When the delta lands in the analytical target, the write path is the same idempotent upsert used for full loads; for the PostGIS-specific mechanics of that write, see loading GeoDataFrames into PostGIS with GeoPandas. Finally, alert on watermark staleness — a watermark that has not advanced within its expected cadence is the earliest, clearest signal that a source stream has silently stalled, long before downstream consumers notice missing data.

Troubleshooting Reference

Related

• Orchestrating Spatial ETL Pipelines — the reference model this scheduling layer plugs into, covering DAG structure, idempotency, and load targets
• Incremental Spatial Loading with Watermark Timestamps — the focused, copy-pasteable watermark load function behind this guide
• Detecting Dataset Changes with ETag and Last-Modified — HTTP conditional-request change detection for file and portal sources
• Syncing STAC Catalogs with pystac-client — datetime-scoped catalog queries that supply the delta for imagery streams
• Managing Spatial Assets with Dagster — partition-based sensors and backfills that express this model natively