Structured Logging for Geospatial Flows

Emit every log line from a spatial pipeline as a single JSON object carrying a fixed schema — task, epsg_code, feature_count, geometry_type, bbox, tile, duration_ms — so throughput, reprojection, and failures become queryable facts rather than free-text prose. This guide shows how to wire structured logging for geospatial flows with structlog and python-json-logger, bind spatial context per flow run, correlate lines to trace IDs, and keep log volume sane inside per-feature loops.

The direct recommendation: standardize on a small, documented set of spatial log fields, bind them once at the start of each Prefect flow or Dagster run with structlog.contextvars, render every record as JSON, and ship the stream to Loki or Elasticsearch. Do this and a single query answers “how many features did the reprojection task push through, per source CRS, in the last hour?” without grepping unstructured text. Structured logging is the narrative layer of observability and monitoring for geospatial pipelines — the counterpart to numeric Prometheus metrics for raster throughput and to request-scoped OpenTelemetry tracing for spatial tasks.

Prerequisites & Architecture Baseline#

Structured logging sits between your task code and your log aggregator. Before wiring it in, pin the libraries and confirm the surrounding pieces are in place. The versions below are the ones this guide’s code was validated against.

• structlog==24.4.0 — the processor pipeline that turns key/value events into JSON.
• python-json-logger==2.0.7 — optional bridge if you route third-party stdlib logs (GDAL, rasterio warnings) through the same JSON formatter.
• geopandas==1.0.1 and rasterio==1.3.10 — the spatial libraries whose operations you are annotating.
• pyproj==3.6.1 — resolves EPSG codes to CRS objects so you log authoritative codes, not guesses.
• prefect==3.0.0 or dagster==1.8.0 — the orchestrator whose run context supplies the flow/run identifiers you bind.
• A log sink that parses JSON: Grafana Loki (with the json pipeline stage) or Elasticsearch 8.x behind a shipper such as Promtail, Fluent Bit, or Vector.
• Clocks synchronized (NTP) across workers so duration_ms and timestamps line up when you correlate logs with traces.

If you have not yet stood up the parent section, read the observability and monitoring overview first — it frames where logs, metrics, traces, and Grafana dashboards for GIS workflows each fit.

Core Principles#

Structured logging for spatial work is not “print, but JSON.” Five principles separate a log stream you can query from one that merely looks tidy.

1. A canonical spatial schema. Agree on one set of field names and stick to them across every task. If one task logs epsg and another logs srid and a third logs crs_code, no query spans them. The schema in the configuration reference below is the contract; treat it as an interface, not a suggestion.

2. Bind context once, inherit everywhere. A flow run has attributes that every line inside it should carry: the run ID, the tile being processed, the source dataset. Bind these once with structlog.contextvars.bind_contextvars and every subsequent log.info(...) in that run — even from nested functions — inherits them automatically. You stop passing context by hand.

3. Events are verbs, fields are nouns. The log event string names what happened (“reprojected features”, “read raster window”) and stays low-cardinality. The variable data — counts, codes, extents — goes into fields. This keeps events groupable and fields filterable. Never interpolate a feature count into the event string.

4. Correlate, don’t duplicate. A log line should carry the trace_id and span_id of the operation it describes so you can pivot from a slow trace to its logs and back. The log records the discrete facts; the trace records the timing and causal tree. Bind the IDs from your OpenTelemetry tracing context rather than logging the same spans twice.

5. Control volume at the source. A loop over 2 million features must not emit 2 million log lines. Aggregate the loop into one summary record, or sample every Nth iteration. Volume control is a first-class design choice, covered under edge cases below.

Production Implementation#

The setup has two parts: a one-time structlog configuration that renders JSON and injects trace IDs, and a small helper that binds spatial context for a flow run. The example instruments a Prefect flow that reprojects a vector layer and writes tiles, but the same configure_logging() and binding pattern drops into a Dagster op unchanged.

import logging
import time
from typing import Any

import geopandas as gpd
import structlog
from opentelemetry import trace
from prefect import flow, task
from pyproj import CRS


def _inject_trace_context(
    logger: logging.Logger, method_name: str, event_dict: dict[str, Any]
) -> dict[str, Any]:
    """structlog processor: attach the active OTel trace/span IDs to every line."""
    span = trace.get_current_span()
    ctx = span.get_span_context() if span else None
    if ctx and ctx.is_valid:
        event_dict["trace_id"] = format(ctx.trace_id, "032x")
        event_dict["span_id"] = format(ctx.span_id, "016x")
    return event_dict


def configure_logging(level: int = logging.INFO) -> None:
    """Render every record as a single JSON object with a stable field order."""
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,     # pull in bound flow context
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            _inject_trace_context,
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer(),          # the output is a JSON line
        ],
        wrapper_class=structlog.make_filtering_bound_logger(level),
        cache_logger_on_first_use=True,
    )


log = structlog.get_logger()


@task
def reproject_layer(path: str, target_epsg: int) -> gpd.GeoDataFrame:
    gdf: gpd.GeoDataFrame = gpd.read_file(path)
    source_crs: CRS = gdf.crs  # authoritative source CRS from the file
    source_epsg: int | None = source_crs.to_epsg() if source_crs else None

    start = time.perf_counter()
    reprojected: gpd.GeoDataFrame = gdf.to_crs(epsg=target_epsg)
    elapsed_ms = round((time.perf_counter() - start) * 1000, 2)

    bounds = reprojected.total_bounds  # (minx, miny, maxx, maxy)
    log.info(
        "reprojected features",
        task="reproject_layer",
        epsg_code=source_epsg,
        target_epsg=target_epsg,
        feature_count=len(reprojected),
        geometry_type=str(reprojected.geom_type.iloc[0]) if len(reprojected) else None,
        bbox=[round(float(b), 6) for b in bounds],
        duration_ms=elapsed_ms,
    )
    return reprojected


@flow(name="tile-reprojection")
def reproject_flow(path: str, target_epsg: int = 3857) -> None:
    configure_logging()
    # Bind run-scoped context ONCE; every line below inherits it.
    from prefect.runtime import flow_run

    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        flow_run_id=str(flow_run.get_id()),
        source_dataset=path,
    )
    log.info("flow started", task="reproject_flow", target_epsg=target_epsg)
    reproject_layer(path, target_epsg)
    log.info("flow finished", task="reproject_flow")

A single line from reproject_layer renders like this — one object, every field queryable, trace IDs present because the flow ran inside an active span:

{
  "event": "reprojected features",
  "level": "info",
  "timestamp": "2026-07-12T09:14:22.517Z",
  "flow_run_id": "b1e7c0d2-9a4f-4c11-8f3a-1d2e3f4a5b6c",
  "source_dataset": "s3://tiles/parcels.gpkg",
  "task": "reproject_layer",
  "epsg_code": 4326,
  "target_epsg": 3857,
  "feature_count": 48211,
  "geometry_type": "Polygon",
  "bbox": [-122.51, 37.70, -122.36, 37.83],
  "duration_ms": 1843.11,
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7"
}

Step-by-Step Walkthrough#

The code above encodes each principle. Read it in the order the log record is built.

1. configure_logging() establishes the pipeline. The processors list runs top to bottom for every record. merge_contextvars runs first so bound run context is present before anything else; JSONRenderer runs last so the final output is always a JSON line. Configure once per worker process, not per task — cache_logger_on_first_use=True makes repeat get_logger() calls cheap.

2. _inject_trace_context correlates logs to traces. It reads the active OpenTelemetry span and writes trace_id/span_id as lowercase hex — the exact format Tempo, Jaeger, and Loki’s trace-to-logs pivot expect. Because it is a processor, it fires on every line automatically; you never pass IDs by hand. This is the concrete mechanism behind principle 4.

3. Binding happens once, in the flow. bind_contextvars(flow_run_id=..., source_dataset=...) stores context in a context-local that merge_contextvars reads back. Every log.info in the flow and in the tasks it calls inherits those two fields. clear_contextvars() at the top guards against leakage if a worker reuses a thread across runs.

4. The task reads the authoritative source CRS. gdf.crs.to_epsg() asks pyproj for the real EPSG code from the file rather than assuming one. Logging epsg_code=None when a file has no CRS is itself a signal — it surfaces the exact undefined-CRS bug that silently corrupts a reprojection. This is why principle 1 insists the field is always present.

5. Timing wraps only the operation. time.perf_counter() brackets the to_crs call, and the result becomes duration_ms. Keeping the window tight means the number reflects the reprojection, not file I/O, so it correlates cleanly with the same operation’s span duration.

6. The event string stays constant. Every reprojection logs the event "reprojected features" regardless of how many features or which CRS. That constancy is what lets you count_over_time the event in Loki or aggregate it in Elasticsearch — the variable data lives in fields, exactly as principle 3 requires. For the narrow, verified version of this pattern, see the companion how-to on logging feature counts and EPSG codes per task.

Edge Cases & Failure Recovery#

Spatial workloads break structured logging in ways generic web services do not — mainly through cardinality and geometry-specific nulls. Here are the cases that bite in production and how to detect and fix each.

Per-feature loops flood the sink#

A task that logs inside a for feature in gdf.iterrows() loop over a national parcel dataset emits millions of lines, blows past Loki’s ingestion limits, and buries the useful summary. Detect it by watching your shipper’s dropped-line counter or Loki’s rate limited errors. Fix it by aggregating: accumulate counts in local variables and emit one summary record after the loop.

import structlog

log = structlog.get_logger()


def clean_geometries(gdf) -> int:
    repaired = 0
    invalid = 0
    for geom in gdf.geometry:
        if not geom.is_valid:
            invalid += 1
            repaired += 1  # buffer(0) or make_valid applied elsewhere
    # ONE line for the whole loop, not one per feature.
    log.info(
        "geometry validation summarized",
        task="clean_geometries",
        feature_count=len(gdf),
        invalid_count=invalid,
        repaired_count=repaired,
    )
    return repaired

When you genuinely need per-feature visibility for debugging, sample it: if i % 1000 == 0: log.debug(...) at DEBUG level, kept out of production by the filtering wrapper’s level.

High-cardinality fields explode the index#

Elasticsearch maps every distinct field key; a per-feature feature_id or a raw floating-point bbox used as a label rather than a value bloats the index and slows queries. Detect it via Elasticsearch’s field-mapping count climbing without bound. Fix it by keeping identifiers as values inside a known field, rounding bbox coordinates (as the code does with round(..., 6)), and never promoting geometry WKT into a searchable field — log a hash or a tile index instead.

Missing or undefined CRS logs as null#

A file with no .prj sidecar yields gdf.crs is None, so epsg_code logs as null. That is correct behavior, but a downstream query filtering epsg_code=4326 silently skips those rows. Detect it with a scheduled query for epsg_code: null and alert when the count is nonzero — the same signal that feeds alerting on CRS validation failures in Grafana. Fix the data upstream; do not paper over it by defaulting the log field.

Non-serializable values crash the renderer#

JSONRenderer raises on objects it cannot encode — a raw Shapely geometry, a NumPy float32, a datetime without a serializer. Detect it as a logging exception in worker stderr. Fix it by coercing at the call site: float(value), int(count), str(geom.geom_type), and lists of Python floats for bbox, exactly as the production code does.

Exceptions lose their spatial context#

A task that raises inside a try without logging first loses the epsg_code/tile context that would explain the failure. Fix it by binding context early and logging with log.exception(...) in the handler; the format_exc_info processor serializes the traceback into the same JSON line, and the bound run context travels with it.

Configuration Reference#

This is the canonical spatial log schema — the contract every task honors. Field names are lowercase snake_case, types are the JSON types after serialization, and “always present?” marks the fields a query is allowed to assume exist on operational log lines.

Keep this table in your repository next to the logging module and treat any new field as an API change. Loki’s json stage and an Elasticsearch index template should both be generated from it so the schema is enforced at the sink, not just in code. For the pipeline stage that parses these lines in Loki, see the official Loki JSON stage documentation.

Frequently Asked Questions#