Observability and Lineage for Spatial Validation

A spatial validation pipeline that produces correct results but cannot explain itself is a liability. When a downstream analyst asks why a parcel layer lost 4% of its features overnight, or when a compliance auditor asks which rule version approved a boundary dataset, “the job ran green” is not an answer. Observability and lineage close that gap: observability tells GIS analysts, QA engineers, and platform teams whether the pipeline is healthy and fast right now, while lineage gives data stewards and compliance officers a durable, reproducible record of which rule ran against which dataset version. This guide sits within the broader Validation Pipeline Architecture and covers how to instrument the rule-evaluation and routing stages so every quality outcome is traceable end to end.

Prerequisites

Confirm each item before instrumenting. Observability retrofitted onto an already-broken pipeline mostly produces noise; the signals below are only meaningful once the validation stages themselves are deterministic.

1. Python 3.10+ with pinned instrumentation libraries. Pin opentelemetry-sdk==1.27.*, opentelemetry-api==1.27.*, opentelemetry-exporter-otlp==1.27.*, openlineage-python==1.19.*, and structlog==24.*. Mismatched OpenTelemetry API and SDK versions silently drop spans rather than raising.

2. A stable run identifier. Every validation execution must mint one run_id (a UUID) at entry and thread it through logs, metric labels, trace attributes, and lineage events. Without a shared correlation key, the four signal types cannot be joined and the unified dashboard is impossible.

3. Versioned rules and datasets. Lineage is only useful if rule_version and dataset_version are real, resolvable values. Tag rule definitions with a semantic version in source control, and stamp each dataset snapshot with a monotonic version or content hash before validation runs.

4. A severity contract already in place. The metrics in this guide are scoped to the blocker, warning, and informational tiers produced by the routing stage. If your pipeline does not yet classify failures, implement Categorizing and Prioritizing Spatial Errors first — severity-scoped metrics depend on it.

5. Collector endpoints reachable. Have an OpenTelemetry Collector (or a compatible backend such as Grafana Tempo/Prometheus) and an OpenLineage endpoint (Marquez, or an OpenLineage-compatible catalog) reachable from the pipeline host. Verify connectivity before wiring exporters, or spans and events queue in memory and are lost on process exit.

Conceptual Foundation

Observability rests on three complementary signal types, often called the three pillars of telemetry: logs (discrete, timestamped records of what happened), metrics (numeric aggregates sampled over time), and traces (causal chains of timed spans across a request or job). Data lineage is a fourth, distinct signal: a provenance graph of datasets and the jobs that transform them. The first three answer operational questions in the moment; lineage answers reproducibility and impact questions over the long term.

For a spatial validation pipeline these map cleanly onto the stages you already run:

Structured logs replace human-readable log lines with machine-parseable records. Instead of INFO validated 12000 parcels, you emit a JSON object carrying run_id, rule_name, rule_version, dataset_version, feature_count, fail_count, severity, and elapsed_ms. The value is queryability: you can ask your log store for every run where valid_topologies failed more than 100 features in the last week, without regex-scraping prose.

Metrics aggregate those per-run facts into time series. The three that matter most for validation are throughput (features validated per second), error rate segmented by severity, and remediation success rate (the fraction of auto-repaired features that pass re-validation). Each is cheap to compute and each answers a different operational question — is the pipeline keeping up, is quality degrading, and is auto-remediation actually working?

Traces decompose a single validation run into timed spans — ingestion, coordinate reference system (CRS) normalisation, rule execution, severity routing — under one trace context. When a nightly run that normally finishes in four minutes suddenly takes forty, a trace shows immediately whether the time went to a slow spatial join, a stalled reprojection, or backpressure in the routing queue.

Lineage records that a validation job consumed dataset parcels@v2026-07-09 and produced parcels_validated@v2026-07-10, using rule set version 1.4.2, and that the run observed a validity rate of 0.981 over 412,300 features. This is provenance, not monitoring. Its consumers are auditors reconstructing a decision months later and stewards performing impact analysis before a schema change.

Two open standards make this vendor-neutral. OpenTelemetry (OTel) is the standard for logs, metrics, and traces — a single instrumentation API with pluggable exporters, so you are not locked to one monitoring vendor. OpenLineage is the standard for lineage events, with a JSON event model of runs, jobs, datasets, and extensible facets. Adopting both means the instrumentation you write today survives a backend migration tomorrow, and it is the reason the child guide Tracking Spatial Data Lineage with OpenLineage can integrate with any OpenLineage-compatible catalog rather than a proprietary one.

Step-by-Step Implementation

Step 1: Emit Structured Logs Per Rule Execution

Configure a structured logger once at pipeline entry, then bind the run-scoped context so every subsequent log line carries it automatically.

import uuid
import time
import logging
import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
)

log = structlog.get_logger()

def new_run_context(dataset_version: str, rule_set_version: str) -> str:
    """Mint a run_id and bind it plus version metadata to all subsequent logs."""
    run_id = str(uuid.uuid4())
    structlog.contextvars.bind_contextvars(
        run_id=run_id,
        dataset_version=dataset_version,
        rule_set_version=rule_set_version,
    )
    return run_id

def log_rule_result(rule_name: str, rule_version: str, feature_count: int,
                    fail_count: int, severity: str, elapsed_ms: float) -> None:
    log.info(
        "rule_executed",
        rule_name=rule_name,
        rule_version=rule_version,
        feature_count=feature_count,
        fail_count=fail_count,
        pass_count=feature_count - fail_count,
        severity=severity,
        elapsed_ms=round(elapsed_ms, 2),
    )

Verification: run new_run_context("parcels@v1", "[email protected]") then log_rule_result("valid_topologies", "1.2.0", 1000, 7, "blocker", 812.4) and confirm the emitted line is a single JSON object containing both the bound run_id and the rule fields — pipe it through python -c "import json,sys; json.loads(sys.stdin.readline())" to prove it parses.

Step 2: Register Severity-Scoped Metrics

Define three OpenTelemetry instruments. Segment error rate and remediation by severity using metric attributes rather than separate instruments, so a dashboard can slice one series many ways.

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

reader = PeriodicExportingMetricReader(OTLPMetricExporter(), export_interval_millis=15000)
metrics.set_meter_provider(MeterProvider(metric_readers=[reader]))
meter = metrics.get_meter("spatial.validation")

features_validated = meter.create_counter(
    "validation.features.total",
    unit="1",
    description="Features processed, labelled by rule and severity outcome",
)
failures_total = meter.create_counter(
    "validation.failures.total",
    unit="1",
    description="Failed feature-rule pairs, labelled by severity",
)
remediation_attempts = meter.create_counter(
    "validation.remediation.attempts.total", unit="1",
    description="Auto-remediation attempts",
)
remediation_success = meter.create_counter(
    "validation.remediation.success.total", unit="1",
    description="Auto-remediation attempts that passed re-validation",
)

def record_rule_metrics(rule_name: str, severity: str,
                        feature_count: int, fail_count: int) -> None:
    attrs = {"rule": rule_name, "severity": severity}
    features_validated.add(feature_count, attrs)
    failures_total.add(fail_count, attrs)

Verification: after one run, query the metrics backend for rate(validation.failures.total{severity="blocker"}[5m]) and confirm it is non-zero only when blocker-tier rules actually failed. The error rate by severity is then validation.failures.total / validation.features.total grouped by the severity label, and remediation success rate is validation.remediation.success.total / validation.remediation.attempts.total.

Step 3: Add Distributed Tracing Spans

Wrap each pipeline stage in a span under one root span per run. Stamp the run_id on the root span so traces cross-link to logs and lineage.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("spatial.validation")

def run_validation(gdf, rules, run_id, dataset_version):
    with tracer.start_as_current_span("validation.run") as root:
        root.set_attribute("run_id", run_id)
        root.set_attribute("dataset_version", dataset_version)
        root.set_attribute("feature_count", len(gdf))

        with tracer.start_as_current_span("stage.crs_normalise"):
            gdf = gdf.to_crs(4326) if gdf.crs.to_epsg() != 4326 else gdf

        with tracer.start_as_current_span("stage.rule_execution") as rule_span:
            results = []
            for rule in rules:
                start = time.perf_counter()
                passed = rule["check"](gdf)
                elapsed_ms = (time.perf_counter() - start) * 1000
                fail_count = int((~passed).sum())
                record_rule_metrics(rule["name"], rule["severity"], len(gdf), fail_count)
                log_rule_result(rule["name"], rule["version"], len(gdf),
                                fail_count, rule["severity"], elapsed_ms)
                results.append((rule["name"], fail_count))
            rule_span.set_attribute("rules.evaluated", len(rules))

        with tracer.start_as_current_span("stage.routing"):
            pass  # route failures to remediation / steward queue / dashboard
        return results

Verification: trigger one run and open the trace in your backend. Confirm the validation.run root span contains three child spans and that stage.rule_execution accounts for the bulk of wall-clock time on a healthy dataset — if stage.crs_normalise dominates, an unexpected reprojection is happening every run.

Step 4: Emit OpenLineage Run Events

Emit a START event when the job begins and a COMPLETE (or FAIL) event when it ends, naming input and output datasets. Attach a custom facet capturing the spatial characteristics and quality outcome. The child guide covers facet design in depth; this is the minimal wiring.

from openlineage.client import OpenLineageClient
from openlineage.client.event_v2 import RunEvent, RunState, Run, Job, Dataset
from openlineage.client.uuid import generate_new_uuid
from datetime import datetime, timezone

ol_client = OpenLineageClient.from_environment()  # reads OPENLINEAGE_URL
PRODUCER = "https://spatial-data-validation.com/validation-pipeline"

def emit_lineage(run_id, state, inputs, outputs, facets=None):
    event = RunEvent(
        eventType=state,
        eventTime=datetime.now(timezone.utc).isoformat(),
        run=Run(runId=run_id, facets=facets or {}),
        job=Job(namespace="spatial-validation", name="parcel_validation"),
        inputs=inputs,
        outputs=outputs,
        producer=PRODUCER,
    )
    ol_client.emit(event)

# On start:
emit_lineage(
    run_id, RunState.START,
    inputs=[Dataset(namespace="gpkg", name="parcels")],
    outputs=[],
)
# On completion, with a custom quality facet:
emit_lineage(
    run_id, RunState.COMPLETE,
    inputs=[Dataset(namespace="gpkg", name="parcels")],
    outputs=[Dataset(namespace="gpkg", name="parcels_validated")],
    facets={"spatialQuality": {
        "_producer": PRODUCER,
        "_schemaURL": PRODUCER + "/facets/spatial-quality.json",
        "crs": "EPSG:4326",
        "featureCount": 412300,
        "validityRate": 0.981,
    }},
)

Verification: open Marquez (or your OpenLineage catalog) and confirm the parcel_validation job shows one completed run whose lineage graph links parcels to parcels_validated, with the spatialQuality facet visible on the run. A run stuck in START with no COMPLETE means the process died mid-validation — itself a useful alerting signal.

Step 5: Wire Alerting Thresholds

Alert on the severity-scoped metrics, not on raw log volume. Define thresholds that reflect the operational cost of each tier: tight on blockers, loose on informational.

-- Alert rule expressed as a PromQL-style condition (evaluated by the metrics backend):

-- Page immediately: blocker error rate exceeds 1% over 10 minutes
sum(rate(validation_failures_total{severity="blocker"}[10m]))
  / sum(rate(validation_features_total{severity="blocker"}[10m])) > 0.01

-- Warn (ticket, no page): remediation success rate drops below 90% over 1 hour
sum(rate(validation_remediation_success_total[1h]))
  / sum(rate(validation_remediation_attempts_total[1h])) < 0.90

-- Warn: throughput collapses to under 500 features/sec over 15 minutes
sum(rate(validation_features_total[15m])) < 500

Verification: deliberately feed a batch with 5% invalid geometries and confirm the blocker alert fires within the evaluation window, while an informational-only degradation does not page. Tune thresholds against a week of baseline data before enabling paging so you calibrate to real variance.

Step 6: Assemble the Dashboard and Correlate

Build one dashboard keyed on run_id. The design goal is single-click pivoting: from a spiking error-rate panel, an operator should reach the exact failing rule version and the dataset version it ran against.

# Pseudocode for the correlation the dashboard performs — join the four signals on run_id.
def correlate(run_id, log_store, metric_store, trace_store, lineage_store):
    logs = log_store.query(run_id=run_id)                 # per-rule JSON records
    metrics = metric_store.series(labels={"run_id": run_id})
    trace = trace_store.get_trace(run_id=run_id)          # span waterfall
    lineage = lineage_store.get_run(run_id=run_id)        # datasets + facets
    return {
        "rule_set_version": lineage.run.facets["nominalTime"] and logs[0]["rule_set_version"],
        "dataset_version": logs[0]["dataset_version"],
        "validity_rate": lineage.outputs[0].facets["spatialQuality"]["validityRate"],
        "slowest_stage": max(trace.spans, key=lambda s: s.duration).name,
        "failing_rules": [r for r in logs if r["fail_count"] > 0],
    }

Verification: pick any run_id from the last day and confirm correlate() returns a coherent object — the failing rules from logs, the validity rate from lineage, and the slowest stage from the trace all describe the same execution. If the validity rate from lineage disagrees with the pass counts in the logs, your facet computation and your metrics are reading different data and must be reconciled.

Common Failure Modes & Fixes

Performance & Scale Considerations

Instrumentation must not dominate the workload. Structured logging at one record per rule per run is negligible, but do not log per feature — a 400,000-feature dataset would emit 400,000 records per rule and swamp both the pipeline and the log store. Aggregate to per-rule counts and reserve per-feature detail for a sampled debug channel or the quarantine queue.

Use batching exporters. BatchSpanProcessor and PeriodicExportingMetricReader buffer and flush on an interval rather than blocking on every span or measurement. Synchronous per-span export can add tens of milliseconds of network latency to every stage, which for a rule loop over many partitions compounds badly. Keep the export interval at 10–30 seconds for validation batch jobs.

Sample traces on high-frequency pipelines. For streaming validation that runs thousands of small jobs per hour, exporting a full trace for every run is expensive and rarely inspected. Apply tail-based or ratio sampling (for example, keep all traces that contain an error span plus 5% of clean runs) so you retain the diagnostically valuable traces without the volume.

Lineage events are cheap; keep them coarse. Emit lineage at job granularity — one START and one COMPLETE per validation run — not per rule or per feature. The provenance graph’s value is in dataset-to-dataset edges, and per-rule lineage explodes the graph without adding reproducibility. Attach per-rule detail as facet fields on the single run event instead.

Decouple emission from the critical path. For pipelines built on the queue-based patterns in Asynchronous Validation Workflows, emit telemetry from the worker after it acknowledges the message, or push events to a local buffer that a sidecar drains. A slow or unreachable observability backend must never block or fail validation itself — instrumentation is best-effort and must degrade gracefully.

Integration with the Validation Pipeline

Observability and lineage are cross-cutting concerns: they wrap every stage of the validation directed acyclic graph (DAG) rather than occupying one node. In practice they attach at three seams.

At rule execution. The rule engine emits one structured log record and one set of metric increments per rule, and runs inside a stage.rule_execution span. This is the richest source of quality signal — a sudden spike in failures on a previously stable rule is the earliest indicator of upstream schema drift or corrupted source data, and severity-scoped metrics surface it before it reaches a downstream map or model.

At routing. The severity router is where remediation-success-rate metrics originate: when an auto-remediation step repairs a geometry and re-validation passes, increment validation.remediation.success.total. Routing decisions themselves belong in the trace, so an operator can see whether a run’s time went to remediation retries. The severity tiers here are the same ones defined in Categorizing and Prioritizing Spatial Errors, which is why the metric labels reuse that taxonomy verbatim.

At job boundaries. Lineage events fire at the edges of the whole job — one START when the validation run picks up a dataset version, one COMPLETE when it produces the validated output. This is the seam that matters most to compliance officers: the lineage record ties a specific rule-set version to a specific dataset version and its measured validity rate, which is precisely the evidence that regulatory frameworks demand. Compliance Framework Alignment explains how that provenance record satisfies audit-trail requirements under standards such as INSPIRE (Infrastructure for Spatial Information in Europe) and ISO (International Organization for Standardization) 19157, and the hands-on emission of those events is the subject of Tracking Spatial Data Lineage with OpenLineage.

The unifying discipline across all three seams is the shared run_id. Because logs, metrics, traces, and lineage all carry it, an operator who notices a failing metric can pivot to the trace that shows where time went, the logs that name the failing rule and its version, and the lineage record that names the dataset version — a complete, reproducible account of one validation run assembled from four independent signals.

Frequently Asked Questions

Related

• Tracking Spatial Data Lineage with OpenLineage — hands-on emission of run events and custom spatial facets into Marquez
• Categorizing and Prioritizing Spatial Errors — the severity taxonomy that scopes the metrics on this page
• Asynchronous Validation Workflows — how to emit telemetry from queue workers without blocking validation
• Compliance Framework Alignment — how lineage records satisfy audit-trail requirements under INSPIRE and ISO 19157
• Building Rule Engines with GeoPandas — the rule-execution stage that emits most of the quality signal

Back to Validation Pipeline Architecture