This guide is part of Validating Spatial Data Quality in Pipelines, within the broader Automated Vector & Raster Cleaning Workflows reference.

Quarantining Invalid Features to a Dead-Letter Table

When a validation gate rejects a row, three things can happen to it: it is dropped, it takes the whole batch down with it, or it is set aside somewhere it can be examined and replayed. Only the third is operable.

Why Dropping Rejected Rows Is the Expensive Option

• A missing feature is invisible. Nobody notices the parcel that was silently discarded until someone asks why a district’s totals are short.
• The failure cannot be diagnosed later. Without the offending geometry, an investigation has only a count, and a count does not explain a cause.
• Fixes cannot be verified. A repair rule written after the fact needs the original input to prove that it works.
• Aborting instead is worse at scale. Failing a hundred-thousand-row batch because eight rows are malformed converts a small data problem into a pipeline outage.

Version and Environment Compatibility

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

What a Dead-Letter Row Has to Carry

The quarantine_features Recipe

from __future__ import annotations

import json
import logging
from datetime import datetime, timezone
from pathlib import Path

import geopandas as gpd
import pandas as pd
import shapely

logger = logging.getLogger(__name__)


def quarantine_features(
    failures: gpd.GeoDataFrame,
    dead_letter_root: Path | str,
    run_id: str,
    source_uri: str,
    layer_name: str,
    code_version: str,
    id_column: str = "parcel_id",
    reason_column: str = "failed_checks",
) -> int:
    """Write rejected features to a partitioned dead-letter dataset.

    The geometry is stored as WKB rather than as a geometry column, because a
    quarantine table must be able to hold geometry that no reader can parse.
    """
    if failures.empty:
        logger.info("no features quarantined for run %s", run_id)
        return 0

    geoms = failures.geometry.values
    attribute_columns = [c for c in failures.columns
                         if c not in {failures.geometry.name, reason_column}]

    records = pd.DataFrame({
        "feature_id": failures[id_column].astype(str).to_numpy(),
        "layer_name": layer_name,
        "source_uri": source_uri,
        "geometry_wkb": shapely.to_wkb(geoms, include_srid=False),
        "source_crs": str(failures.crs),
        "attributes_json": [
            json.dumps({k: _jsonable(v) for k, v in row.items()})
            for row in failures[attribute_columns].to_dict("records")
        ],
        "failed_checks": failures.get(reason_column, "unspecified"),
        "validity_message": [
            shapely.is_valid_reason(geom) if geom is not None else "missing geometry"
            for geom in geoms
        ],
        "run_id": run_id,
        "code_version": code_version,
        "quarantined_at": datetime.now(timezone.utc),
    })
    records["quarantine_date"] = records["quarantined_at"].dt.date.astype(str)

    records.to_parquet(
        Path(dead_letter_root),
        partition_cols=["layer_name", "quarantine_date"],
        index=False,
    )
    logger.warning("quarantined %d features from %s (run %s)", len(records), layer_name, run_id)
    return len(records)


def _jsonable(value):
    """Coerce pandas and numpy scalars into something json.dumps accepts."""
    if pd.isna(value):
        return None
    if hasattr(value, "isoformat"):
        return value.isoformat()
    if hasattr(value, "item"):
        return value.item()
    return value

Key Implementation Notes

• Geometry is stored as WKB in a binary column, not as a geometry column. A quarantine table has to accept geometry that a strict reader would reject, and a typed geometry column can refuse the write — which loses exactly the row you most needed.
• is_valid_reason is captured per row. The GEOS message is the difference between “380 invalid geometries” and “380 ring self-intersections at one shared boundary”, and only the second suggests a cause.
• Attributes are serialised to JSON rather than kept as columns. Source schemas differ and drift; one JSON column absorbs that without the quarantine table needing a migration every time an upstream layer changes.
• Partitioning is by layer and quarantine date. Expiry becomes a partition drop, and a replay for one layer on one day reads one directory.
• The function returns a count so the caller can compute the rate and publish it as a metric — the rate, not the count, is what a threshold should watch.
• Nothing is repaired here. Repair is a separate decision made against the evidence, and mixing it into the quarantine write destroys the evidence.

Turning the Rate Into a Signal

The count of quarantined rows is nearly meaningless on its own, because batch sizes vary. The rate is the number worth watching, and it deserves three pieces of treatment.

Publish it as a metric, not a log line, with the layer and the failing check as labels. That makes “which check is driving the increase?” a query rather than an investigation.

Threshold it against a measured baseline. A layer that normally quarantines 0.1% and today quarantines 0.4% is worth a look; the same 0.4% on a layer that normally sits at 0.35% is noise. The threshold belongs in the same configuration as the severity mapping.

Alert on the derivative as well as the level. A rate that has doubled every night for three nights will cross any fixed threshold eventually, and catching it on night two rather than night five is the difference between a question for the publisher and a backfill.

Troubleshooting the Quarantine Path

Integration Note

The quarantine write sits immediately after validation, taking the failures frame produced in writing geometry validity assertions with pandera and returning a count the caller turns into a metric. In an orchestrated pipeline the replay is its own scheduled job rather than a step in the main DAG, so that a stuck replay cannot delay the nightly load — the separation argued for in orchestrating spatial ETL pipelines.

Choosing the Sink for Quarantined Rows

Three sinks are common, and the right one depends on who investigates the failures rather than on where the main data lives.

Partitioned Parquet on object storage is the default. It costs nothing to write, holds arbitrary attribute shapes in a JSON column, and a replay reads exactly one partition. Its weakness is interactive investigation: querying it means opening a notebook or pointing an engine at it.

A relational table beside the target suits pipelines whose failures are examined by the same people who query the serving data. Investigation is a SELECT away, and the join back to the main table is trivial. The cost is schema rigidity — a new source with different attributes needs a migration unless the attributes go into a JSON column here too.

A message queue suits streaming ingestion where the replay is itself a consumer. It fits the architecture, but nothing about it supports the question “how many features did this layer quarantine last month”, which is the question most often asked.

Whatever the sink, keep it separate from the serving table. Quarantined rows in the same table behind a status flag inevitably leak into a query that forgot the filter, and a rejected geometry that reaches a dashboard is worse than one that never arrived.

Related

• Validating Spatial Data Quality in Pipelines
• Writing Geometry Validity Assertions with pandera
• Geometry Repair with Shapely & GeoPandas
• Detecting CRS Drift Between Pipeline Stages