Structured JSON Logging for Geospatial Pipelines

Free-text log lines are hostile to auditing: they cannot be filtered, aggregated, or verified at scale. This guide shows how to configure Python’s standard logging module to emit one machine-parseable JSON object per event, attach a correlation id to every record in a batch, and produce per-batch structured entries that a compliance reviewer can query directly. The result plugs straight into the schema laid out in the parent module, Validation Log Design for Compliance Pipelines.

Prerequisites #

• A geospatial pipeline organized into discrete stages that each produce a GeoDataFrame
• Python 3.9+ with GeoPandas 1.0+ and PyProj 3.4+ installed
• A batch or job identifier available at the entry point of every pipeline run
• Write access to an append-only log sink such as a versioned object store or file handler
• Agreement on the required fields per record with whoever consumes the audit trail

Step-by-step #

Step 1: Build a JSON formatter #

Subclass logging.Formatter so that every emitted line is a complete JSON object. Merge any structured fields passed through the extra argument, and always include a UTC timestamp and the level.

import json
import logging
from datetime import datetime, timezone

RESERVED = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__)

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        # Fold any structured fields passed via logger.info(..., extra={...}).
        for key, value in record.__dict__.items():
            if key not in RESERVED and not key.startswith("_"):
                payload[key] = value
        return json.dumps(payload, sort_keys=True, default=str)

Using sort_keys=True makes each line canonical, which matters when records are later hashed for tamper-evidence.

Step 2: Attach the formatter to a handler #

Wire the formatter into a handler and logger once, at pipeline startup. Disable propagation so records are not double-emitted by the root logger.

def configure_audit_logger(sink_path: str) -> logging.Logger:
    logger = logging.getLogger("compliance.audit")
    logger.setLevel(logging.INFO)
    logger.propagate = False
    handler = logging.FileHandler(sink_path)   # swap for an object-store handler in prod
    handler.setFormatter(JsonFormatter())
    logger.handlers.clear()                    # idempotent: avoid duplicate handlers
    logger.addHandler(handler)
    return logger

Step 3: Propagate a correlation id #

A correlation id ties every record from one run together. Store it in a contextvars.ContextVar so it survives across function boundaries and worker threads without being threaded through every signature manually. A logging.Filter injects it into each record automatically.

import contextvars
import uuid

batch_id_var = contextvars.ContextVar("batch_id", default="unset")

class CorrelationFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.batch_id = batch_id_var.get()   # stamped onto every emitted record
        return True

def start_batch() -> str:
    batch_id = uuid.uuid4().hex
    batch_id_var.set(batch_id)
    return batch_id

Add the filter to the handler with handler.addFilter(CorrelationFilter()). Every subsequent record now carries batch_id with no extra arguments at the call site.

Step 4: Emit a per-batch structured record #

At the end of each stage, emit a single aggregate record. Always read the CRS from the data and project to a metric CRS before any distance summary, so logged metrics are measured in meters rather than degrees.

import geopandas as gpd

def log_stage_summary(logger: logging.Logger, stage: str,
                      gdf: gpd.GeoDataFrame, metric_crs: str = "EPSG:32610") -> None:
    # Project to a linear CRS before summarizing any area metric.
    projected = gdf.to_crs(metric_crs) if gdf.crs else gdf
    logger.info(
        "stage complete",
        extra={
            "stage": stage,
            "epsg_in": gdf.crs.to_epsg() if gdf.crs else None,
            "epsg_metric": 32610,
            "feature_count": int(len(gdf)),
            "empty_geometry_count": int(projected.geometry.is_empty.sum()),
            "total_area_sqm": float(projected.geometry.area.sum()),
        },
    )

Verification #

Confirm that each emitted line is valid JSON, carries the correlation id, and shares one batch_id across the run. Parse the sink back and assert structure rather than eyeballing it.

import json

with open("audit.log") as fh:
    records = [json.loads(line) for line in fh if line.strip()]

batch_ids = {r["batch_id"] for r in records}
assert len(batch_ids) == 1, f"expected one batch id, saw {batch_ids}"
assert all("stage" in r and "feature_count" in r for r in records)
print(f"{len(records)} records verified for batch {batch_ids.pop()}")

If every record parses and the assertions hold, the log is queryable and ready for the hash-chaining and retention steps described in the parent module.

Common Pitfalls #

• Logging geometry objects directly. Shapely geometries are not JSON-serializable and will bloat the log if coerced to WKT. Log counts, areas, and identifiers instead of raw geometry; the default=str fallback should be a safety net, not the plan.
• Summarizing area in a geographic CRS. Calling .area on EPSG:4326 data yields square degrees, which are meaningless. Always reproject to a metric CRS first, as in Step 4, and record which EPSG the metric was computed in.
• Duplicate handlers on re-import. Calling the configuration function twice stacks handlers and emits every line multiple times. Clear existing handlers first, as shown, so the logger is idempotent.

Fields Every Event Carries #

Structured logging pays off only when the structure is consistent, and consistency comes from a small set of fields present on every event regardless of type.

A timestamp in UTC, in a fixed format, so ordering and range queries work without parsing ambiguity. A run identifier tying every event of a run together. An event type from a closed vocabulary, so a reader can dispatch. A schema version, so old records stay readable when the shape changes. And a severity, kept honest: informational for the normal course, warning for something a human should see eventually, error for something that stopped work.

Everything else is type-specific and belongs in a nested object rather than at the top level, which keeps the common fields easy to index and the type-specific ones easy to evolve.

def event(kind: str, run_id: str, severity: str = "info", **payload) -> dict:
    """One shape for every event: common fields flat, type-specific fields nested."""
    return {
        "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
        "run_id": run_id,
        "event": kind,                 # from a closed vocabulary
        "schema": EVENT_SCHEMA,
        "severity": severity,
        "data": payload,               # everything type-specific lives here
    }

Avoiding the Two Common Failure Modes #

Structured logs fail in two opposite directions, and both are avoidable with a rule rather than with discipline.

Free text smuggled into structure. An event whose only useful content is a formatted message string is a text log with JSON overhead. The fix is to require that anything a query might filter on is its own field: parcel identifier, rule identifier, counts, reasons as codes rather than sentences. A human-readable message can accompany them, but it must not be the only place the information exists.

Unbounded cardinality. Emitting an event per parcel per stage produces tens of millions of records for a county run, most carrying nothing that varies. The fix is the granularity rule from lineage work: layer-level for anything applied uniformly, feature-level only for things that genuinely differ. The verdict events are the legitimate high-cardinality stream, and they are the ones worth paying to keep.

Both failure modes are cheap to catch in review: a new event type should be accompanied by the query someone will run against it. If no such query exists, the event probably should not either.

Where the Records Go #

The destination shapes what the log can be used for, and compliance logs have different requirements from application logs.

Newline-delimited JSON to object storage, partitioned by run and date, is the pragmatic default: cheap, durable, immune to schema migration, and directly queryable by most analytical tools. Rolling completed partitions into Parquet gives fast columnar queries over history without changing what was written.

What to avoid for the compliance stream is a hosted logging service billed by ingest volume, since the cost pressure eventually reduces what gets captured — and the fields dropped are usually the ones that seemed redundant until they were needed. Keep operational logging there if it helps, and keep the compliance stream where its retention is under your control.

Related #

Part of: Validation log design

• Capturing CRS provenance in validation logs — the frame facts these events carry.
• Redacting sensitive parcel data from logs — what must not reach the sink.
• Tracking data lineage across geospatial ETL steps — the events that describe transformations.
• Building interactive HTML compliance dashboards — the view built over these records.

Frequently Asked Questions #

Why use the standard logging module instead of just writing JSON files myself? #

The standard library gives you handler routing, log levels, thread-safety, and filters for free. A correlation-id filter stamps every record automatically, and swapping a file sink for an object-store or syslog handler becomes a one-line change rather than a rewrite. Rolling your own file writer usually reinvents these features less reliably.

How do correlation ids work across multiprocessing worker pools? #

A contextvars.ContextVar is inherited by child threads but not by separate processes. When you fan out with multiprocessing, pass the batch id explicitly to each worker and call the setter inside the worker before logging. Each process then stamps the same id, so records from all workers still collate into one run in the audit store.

Should I log every feature or only aggregates? #

Log per-batch aggregates by default and reserve per-feature records for failures and quarantined geometries. At municipal scale, per-feature logging inflates storage and slows queries, while aggregate records plus targeted failure entries give auditors everything they need without the volume.

How does this connect to capturing coordinate system details? #

The formatter here carries whatever fields you pass in extra, including EPSG codes. To record the full source-to-target transformation pipeline rather than just a single EPSG code, follow Capturing CRS Provenance in Validation Logs, which extends these same structured records with pyproj pipeline detail.