This guide is part of Monitoring & Observability for Spatial Pipelines, within the broader Orchestrating Spatial ETL Pipelines reference.

Emitting Metrics for Spatial Record Counts

Counting rows sounds trivial and is the instrumentation most pipelines get subtly wrong — usually by counting in the wrong place, or in a way that a retry corrupts.

Why Naive Counting Misleads

• Counting inside the transform double-counts on retry. A monotonic counter incremented mid-task adds again on the second attempt, so a routine retry looks like a traffic spike.
• Counting only the output hides the loss. Ten thousand rows written is meaningless without knowing whether twelve thousand or ten thousand arrived.
• Absolute counts cannot be thresholded. Batch sizes vary; a rule that fires below 5 000 rows is wrong the first week the source publishes a small update.
• Per-tile labels destroy the metrics store. Four hundred tiles times five metrics times a year of runs is millions of series for information nobody queries that way.

Version and Environment Compatibility

pip install "geopandas>=1.0" "shapely>=2.0"

Where the Counters Belong

The stage_counts Recipe

from __future__ import annotations

import logging
from dataclasses import dataclass, asdict

import geopandas as gpd
import shapely

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class StageCounts:
    dataset: str
    stage: str
    run_id: str
    rows_in: int
    rows_out: int
    rows_quarantined: int
    invalid_geometries: int
    empty_geometries: int
    total_area_m2: float
    area_crs: str

    @property
    def retention(self) -> float:
        return self.rows_out / self.rows_in if self.rows_in else 0.0

    @property
    def quarantine_rate(self) -> float:
        return self.rows_quarantined / self.rows_in if self.rows_in else 0.0


def stage_counts(
    before: gpd.GeoDataFrame,
    after: gpd.GeoDataFrame,
    quarantined: int,
    dataset: str,
    stage: str,
    run_id: str,
) -> StageCounts:
    """Measure one stage boundary, computing area in a metric CRS."""
    geoms = after.geometry.values
    measure = after if (after.crs and after.crs.is_projected) else after.to_crs(
        after.estimate_utm_crs() if len(after) else 3857
    )

    counts = StageCounts(
        dataset=dataset,
        stage=stage,
        run_id=run_id,
        rows_in=len(before),
        rows_out=len(after),
        rows_quarantined=quarantined,
        invalid_geometries=int((~shapely.is_valid(geoms)).sum()),
        empty_geometries=int(shapely.is_empty(geoms).sum()),
        total_area_m2=float(measure.geometry.area.sum()),
        area_crs=str(measure.crs),
    )
    logger.info("%s/%s: %d → %d (retention %.3f, quarantine %.4f)",
                dataset, stage, counts.rows_in, counts.rows_out,
                counts.retention, counts.quarantine_rate)
    return counts


def publish(counts: StageCounts, sinks: list) -> None:
    """Fan the measurement out; a sink failure must never fail the run."""
    payload = asdict(counts) | {
        "retention": counts.retention,
        "quarantine_rate": counts.quarantine_rate,
    }
    for sink in sinks:
        try:
            sink.upsert(key=(counts.dataset, counts.stage, counts.run_id), values=payload)
        except Exception:  # noqa: BLE001 — telemetry is best effort by design
            logger.exception("metrics sink %s failed", type(sink).__name__)

Key Implementation Notes

• upsert keyed on run and stage, not increment. A retried task rewrites its measurement rather than adding to it, so retries never produce phantom spikes.
• Ratios are derived at emission, not at query time. Storing retention alongside the raw counts means an alert rule does not depend on two series arriving together.
• Area is computed in a metric CRS and the CRS is recorded. A series that silently switched from UTM to Web Mercator would show a step change that looks like a data problem and is not.
• Validity and emptiness are counted after the stage. They describe what this stage produced, which is the quantity a stage owner can act on.
• The sink interface is two methods wide. Keeping it minimal is what allows the same call sites to write to the orchestrator’s metadata, a metrics store and a JSON file beside the output.
• Zero-row batches yield zero ratios rather than raising. An empty batch is a legitimate state that the alerting layer, not the counter, should judge.

Labels, Cardinality and What to Keep Elsewhere

Every label multiplies the series count, and spatial pipelines have unusually tempting high-cardinality dimensions. Tile identifier, scene id, feature id and source URI all feel like natural labels and all of them are traps.

The workable label set is small: dataset, stage, and optionally source where a pipeline reads from a handful of named providers. That is a few dozen series per metric, which any store handles indefinitely.

Partition-level detail belongs in the orchestrator’s run metadata, where it is stored once per run and queried by run rather than by series. Feature-level detail belongs in the quarantine table. Neither belongs in a metrics label, and the discipline is easiest to hold if the emitter simply does not accept arbitrary labels — a fixed signature is a better guard than a convention.

Alerting on Ratios Rather Than Counts

The counts exist so that ratios can be computed, and it is the ratios that carry the alerting rules.

Retention below its normal band means a transform is dropping rows it did not use to drop. This catches a filter whose predicate stopped matching after a schema change, which produces no error anywhere.

Quarantine rate above its threshold means the input got worse. Trend matters more than level here, as described in validating spatial data quality in pipelines.

Area per row is the most spatially specific of the three, and the most sensitive to a reprojection: an unintended CRS change moves it by orders of magnitude while leaving every row count untouched.

Each rule needs a floor on the batch size to avoid firing on tiny updates, where a single quarantined row is 20% of the batch.

Troubleshooting Metric Emission

Integration Note

Call stage_counts at each stage boundary and publish once at the end of the task, so a single failure path covers all sinks. The counts also belong in the orchestrator’s own metadata — Dagster asset metadata, Airflow task instance fields, Prefect artifacts — which is where an operator investigating one run will look first, as described in monitoring and observability for spatial pipelines.

Backfilling the Metric History

A new metric starts with no history, which means no baseline and therefore no threshold. Waiting a month to find out what normal looks like is the honest approach and rarely the practical one.

Where the pipeline writes its outputs to durable storage, the history can often be reconstructed. Row counts come from reading the row group metadata of each historical Parquet file — cheap, since it needs no data pages. Coverage area needs the geometries, so it is more expensive, but a sample per historical partition is usually enough to establish a range. Quarantine rates generally cannot be reconstructed, because the rejected rows were never written; that metric has to accumulate forward.

Backfilled points should be marked as such, with a flag or a separate series name. A reconstructed count is not identical to a measured one — it reflects what survived rather than what arrived — and a threshold derived from a mixed series will be subtly wrong in the direction that matters.

Related

• Monitoring & Observability for Spatial Pipelines
• Structured Logging for Geospatial ETL Jobs
• Alerting on Spatial Data Freshness SLAs
• Validating Spatial Data Quality in Pipelines