Logging Feature Counts and EPSG Codes per Task

Bind feature_count plus the source and target EPSG codes to every log line a Prefect or Dagster task emits, using structlog.contextvars, so a single jq or Loki query can slice throughput and reprojection volume by CRS. The trick is to bind the values once when the task starts — after you have read the layer and resolved its CRS with pyproj — and let structlog’s context merge them into every subsequent line automatically, including the final summary line that reports the count. This how-to gives a self-contained task, the exact jq query to verify it, and the pitfalls that make CRS logging lie.

This is the focused technique behind the broader structured logging for geospatial flows guide, which defines the full spatial log schema. Here we do one thing precisely: make feature_count, epsg_code, and target_epsg reliable, per-task, queryable fields.

When to Use This Pattern

Reach for per-task feature-count and EPSG binding when:

You need to answer “how many features flowed through the reprojection step, grouped by source CRS?” from logs alone.
Reprojection bugs are suspected and you want to see every 4326 → 3857 transform’s volume without instrumenting a metrics backend.
A layer’s CRS is sometimes undefined and you need epsg_code: null to surface as a queryable, alertable fact.
You are debugging a specific run and want to pivot from a slow trace span to the exact feature counts that task processed.
You want the throughput numerator that complements numeric Prometheus metrics for raster throughput without yet standing up Prometheus.

If you instead need the full processor pipeline, JSON rendering setup, and volume-control strategy, start with the parent structured logging guide and return here for the CRS-slicing detail.

Complete Working Example

The task below reads a vector file, resolves both the source and target EPSG codes through pyproj, binds all three fields to the run context, reprojects, and emits one summary line. Because the fields are bound rather than passed, any warning or nested log inside the task also carries the count and CRS — which is what makes the later jq query complete.

import logging
from typing import Any

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


def configure_json_logging() -> None:
    """Emit one JSON object per line, with bound context merged in."""
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        cache_logger_on_first_use=True,
    )


log = structlog.get_logger()


@task
def reproject_and_log(source_path: str, target_epsg: int) -> gpd.GeoDataFrame:
    gdf: gpd.GeoDataFrame = gpd.read_file(source_path)

    # Resolve the AUTHORITATIVE source EPSG from the file's CRS, not a guess.
    source_crs: CRS | None = gdf.crs
    source_epsg: int | None = source_crs.to_epsg() if source_crs is not None else None

    # Bind once: every line from here on inherits these three fields.
    structlog.contextvars.bind_contextvars(
        task="reproject_and_log",
        feature_count=len(gdf),
        epsg_code=source_epsg,        # source CRS (may be null → undefined CRS)
        target_epsg=int(target_epsg), # destination CRS
    )

    if source_epsg is None:
        # Still emitted WITH feature_count + target_epsg bound.
        log.warning("source crs undefined; reprojection may be wrong")

    reprojected: gpd.GeoDataFrame = gdf.to_crs(epsg=target_epsg)
    log.info("reprojection complete")  # inherits all four bound fields
    structlog.contextvars.clear_contextvars()
    return reprojected


@flow(name="crs-throughput")
def crs_throughput_flow(paths: list[str], target_epsg: int = 3857) -> None:
    configure_json_logging()
    for path in paths:
        reproject_and_log(path, target_epsg)

A run over three files produces one JSON line per task invocation. Redirected to a file, the log looks like this:

{"event": "reprojection complete", "level": "info", "timestamp": "2026-07-12T10:02:11.004Z", "task": "reproject_and_log", "feature_count": 48211, "epsg_code": 4326, "target_epsg": 3857}
{"event": "reprojection complete", "level": "info", "timestamp": "2026-07-12T10:02:13.882Z", "task": "reproject_and_log", "feature_count": 1290, "epsg_code": 25831, "target_epsg": 3857}
{"event": "source crs undefined; reprojection may be wrong", "level": "warning", "timestamp": "2026-07-12T10:02:14.771Z", "task": "reproject_and_log", "feature_count": 774, "epsg_code": null, "target_epsg": 3857}

The Dagster equivalent is identical inside the function body — swap @task/@flow for @op/@job and call configure_json_logging() once at process start; the contextvars binding is orchestrator-agnostic because it lives in Python’s contextvars, not the framework.

Parameter & Option Reference

The fields this pattern binds, and the knobs that govern them.

Parameter	Type	Default	Spatial notes
`feature_count`	`int`	`len(gdf)`	Row count after read, before filtering. Bind post-read so it reflects what the task actually processed.
`epsg_code`	`int \| None`	`gdf.crs.to_epsg()`	Source CRS. `None` when the file lacks a CRS — keep it null, do not default it.
`target_epsg`	`int`	required	Destination CRS passed to `to_crs(epsg=...)`. Coerce with `int()` so it serializes cleanly.
`task`	`str`	function name	Group-by dimension; keep it stable across runs.
`merge_contextvars`	processor	enabled	Must be first in the processor list so bound fields appear on every line.
`to_epsg(min_confidence)`	`int`	`70`	pyproj’s match threshold; lower it only if a valid CRS returns `None` unexpectedly.

Verification & Testing

Verify the log stream two ways: a jq slice over the raw JSON, and an assertion in a unit test.

First, slice throughput by CRS pair straight from the log file. This groups every reprojection line by its epsg_code → target_epsg transform and sums the features moved:

jq -rs '
  map(select(.event == "reprojection complete"))
  | group_by([.epsg_code, .target_epsg])
  | map({
      transform: "\(.[0].epsg_code) -> \(.[0].target_epsg)",
      tasks: length,
      features: (map(.feature_count) | add)
    })
' flow.log

Against the sample output above (plus the undefined-CRS line, which has event "source crs undefined..." and is excluded by the select), this returns aggregated throughput per CRS pair:

[
  {"transform": "4326 -> 3857", "tasks": 1, "features": 48211},
  {"transform": "25831 -> 3857", "tasks": 1, "features": 1290}
]

To count files that arrived with an undefined CRS — the alertable case — filter for the null directly:

jq -rs 'map(select(.epsg_code == null)) | length' flow.log

Second, assert the binding in a test by capturing structlog’s output. Use structlog.testing.capture_logs so no JSON parsing is needed:

import geopandas as gpd
from shapely.geometry import Point
from structlog.testing import capture_logs


def test_reproject_binds_crs_and_count() -> None:
    gdf = gpd.GeoDataFrame(
        {"id": [1, 2]},
        geometry=[Point(-122.4, 37.8), Point(-122.3, 37.7)],
        crs="EPSG:4326",
    )
    gdf.to_file("/tmp/pts.gpkg", driver="GPKG")

    with capture_logs() as logs:
        reproject_and_log.fn("/tmp/pts.gpkg", target_epsg=3857)

    done = [line for line in logs if line["event"] == "reprojection complete"][0]
    assert done["feature_count"] == 2
    assert done["epsg_code"] == 4326
    assert done["target_epsg"] == 3857

Calling .fn(...) runs the task body directly without a Prefect engine, so the test stays fast. For end-to-end confidence, run the flow and pipe its output through the jq query in CI, failing the build if the null-CRS count exceeds zero.

Common Pitfalls

Reading epsg_code from a variable instead of the CRS. Always resolve it with gdf.crs.to_epsg() at read time. Hard-coding or inferring the source EPSG defeats the purpose — the log must record what the file actually declared, including null when it declares nothing.
Binding before the read. If you bind_contextvars(feature_count=...) before gpd.read_file, the count is wrong or unavailable. Read first, resolve the CRS, then bind — the order in the example is deliberate.
Forgetting clear_contextvars(). Prefect and Dagster reuse worker threads. Without a clear at task end, the next task inherits the previous run’s feature_count and epsg_code, silently poisoning your jq aggregation. Clear on the way out (or use structlog.contextvars.bound_contextvars(...) as a context manager).
Letting to_epsg() return None for a valid CRS. A custom or compound CRS can fail the default confidence match and log null even though the data is projected. When you see unexpected nulls for known-good data, lower to_epsg(min_confidence=...) or log the WKT-derived authority code, and cross-check against validating coordinate systems before ETL.

Related: This technique feeds the schema defined in structured logging for geospatial flows, complements numeric Prometheus metrics for raster throughput, correlates to spans via OpenTelemetry tracing for spatial tasks, and surfaces on Grafana dashboards for GIS workflows.

← Back to Structured Logging for Geospatial Flows

When to Use This Pattern#

Complete Working Example#

Parameter & Option Reference#

Verification & Testing#

Common Pitfalls#

When to Use This Pattern

Complete Working Example

Parameter & Option Reference

Verification & Testing

Common Pitfalls