#Cost & Observability for Conversion Pipelines

A batch conversion service that turns Shapefiles and GeoJSON into GeoParquet is deceptively easy to stand up and surprisingly hard to run. The happy path — read a file, validate geometry, write a compressed Parquet object — works fine in a demo. In production the same code processes thousands of heterogeneous inputs a day across ephemeral cloud workers, and the questions that matter change entirely. How much does it cost to convert a gigabyte? Which datasets are quietly dropping geometries? Why did last night’s run take three times longer than usual, and did that show up on the bill? Without instrumentation these questions are unanswerable, and the pipeline drifts into a state where nobody can tell a healthy run from a degraded one until a downstream consumer complains.

This guide treats cost and reliability as first-class engineering concerns for the conversion stage, sitting within the broader Data Conversion & Migration Pipelines topic area. The goal is a pipeline that emits enough signal — structured logs, metrics, traces, and a per-job cost record — that you can answer “what did this run cost and did it work” for any dataset, on any day, without re-running it. We will cover the signals worth capturing, where to emit them from, how to derive dollars per gigabyte from raw resource usage, and how to fold everything into service level objectives and dashboards that a small team can actually maintain.

#Prerequisites

• Python 3.10+ with pyarrow>=14.0, geopandas>=0.14, structlog>=24.1, and opentelemetry-sdk>=1.24
• A working conversion job: ideally one built from the patterns in Building Batch Conversion Pipelines with Python, so there is a clear unit of work to instrument
• A metrics backend: Amazon CloudWatch, Prometheus, or any OpenTelemetry-compatible collector, plus somewhere to store per-job cost records (a table, a lake, or object storage)
• Cloud billing familiarity: knowing the compute rate per vCPU-second, the object-storage request and egress rates, and how your scheduler charges for idle time
• Baseline throughput numbers: rows per second and bytes per second for a representative dataset, so anomalies have a reference point

#Architectural Foundations

Observability for a conversion pipeline rests on three complementary signal types, and conflating them is the most common early mistake. Logs are discrete, high-cardinality events — “started dataset X”, “dropped 4 invalid geometries”, “wrote 812 MB”. Metrics are pre-aggregated numeric time series — rows converted per minute, p95 duration, running cost per gigabyte. Traces stitch a single job’s stages into a causal timeline so you can see that the write step, not the read step, is where a slow run spent its time. A mature pipeline emits all three from the same instrumentation point and correlates them with a shared job identifier.

Cost sits on top of these signals as a derived quantity. You almost never measure dollars directly inside a worker; instead you measure the physical resources a job consumed — compute seconds, bytes read, bytes written, request counts — and multiply by unit rates that live in configuration. This separation matters because rates change: a move to a cheaper instance family or a new storage tier should update one config value, not require re-instrumenting every job. It also lets you attribute cost precisely, because the same resource counters that feed the bill also feed per-dataset cost records. The full mechanics of that calculation are covered in Cost-per-GB Tracking for Conversion Pipelines.

The diagram below shows how the three signal types flow out of a conversion job’s stages, through a shared instrumentation layer, and into the collector, dashboards, and alerting that turn raw telemetry into operational answers:

The reason to route every stage through one instrumentation layer rather than sprinkling print statements and ad hoc counters is correlation. When a run degrades, you want to pivot from a metric anomaly (“cost per gigabyte doubled at 02:00”) to the exact jobs responsible (“these 40 datasets retried the write step four times each”) to the log lines and trace spans that explain why — all keyed on the same identifier. That pivot is only possible if the identifier is stamped once, at the top of the job, and carried everywhere.

#Step-by-Step Workflow

#1. Define the Signals and Their Units

Before writing any emitter, enumerate what you will measure and pin its units. Ambiguity here — seconds versus milliseconds, source bytes versus compressed bytes — poisons every dashboard downstream. For a Shapefile-to-GeoParquet job the core set is small: bytes read from source, rows converted, geometries dropped, bytes written, compute seconds, and terminal status. Everything else derives from these.

# python>=3.10
from dataclasses import dataclass

@dataclass(frozen=True)
class ConversionSignals:
    dataset_id: str
    source_bytes: int          # size of the input Shapefile/GeoJSON
    rows_converted: int        # features written to GeoParquet
    geometries_dropped: int    # invalid/null geometries skipped
    output_bytes: int          # size of the written GeoParquet object
    compute_seconds: float     # wall-clock CPU time for the job
    status: str                # "ok" | "partial" | "failed"

Note that source_bytes and output_bytes are both captured: their ratio is your realised compression, and it feeds directly into cost attribution. Choosing an aggressive codec such as those discussed in ZSTD Compression Levels for Geospatial Data shrinks output_bytes and therefore the storage line of every future bill, so the two signals together tell you whether a codec change actually paid off.

#2. Emit Structured Logs with a Correlation ID

Human-readable log strings are useless at scale. Emit JSON objects where every line for a given job shares a correlation ID, so a query backend can reconstruct the full story of one dataset. The mechanics — key naming, correlation ID propagation, and shipping to CloudWatch or a log lake — are the subject of Structured Logging and Metrics for Migration Jobs; here the point is simply that logging happens through a bound context, not free-form strings.

# structlog>=24.1
import structlog

log = structlog.get_logger()

def convert(dataset_id: str, job_id: str) -> None:
    bound = log.bind(job_id=job_id, dataset_id=dataset_id)
    bound.info("conversion.start")
    # ... do work ...
    bound.info("conversion.done", rows=812_004, dropped=4)

#3. Record Metrics and a Per-Job Cost Record

Logs answer “what happened to this dataset”; metrics answer “how is the fleet doing”. Emit counters and histograms for the signals defined in step 1, then, at job end, compute a cost record by multiplying resource usage by unit rates. Keep the rates in configuration so a pricing change is a one-line edit.

# python>=3.10
UNIT_RATES = {
    "compute_usd_per_sec": 0.0000116,   # vCPU-second on the chosen instance
    "storage_usd_per_gb_month": 0.023,  # standard-tier object storage
}

def cost_of(sig: "ConversionSignals") -> float:
    compute = sig.compute_seconds * UNIT_RATES["compute_usd_per_sec"]
    storage = (sig.output_bytes / 1024**3) * UNIT_RATES["storage_usd_per_gb_month"]
    return round(compute + storage, 6)

#4. Trace Across Stages

A single duration metric tells you a job was slow but not where. Wrap each stage in a span so tail latency is attributable. OpenTelemetry spans nest naturally under a parent job span, giving you a flame graph per conversion when you need to debug a slow tail.

# opentelemetry-sdk>=1.24
from opentelemetry import trace

tracer = trace.get_tracer("conversion")

def convert_traced(dataset_id: str) -> None:
    with tracer.start_as_current_span("convert", attributes={"dataset_id": dataset_id}):
        with tracer.start_as_current_span("read"):
            ...  # read Shapefile
        with tracer.start_as_current_span("write"):
            ...  # write GeoParquet

#5. Turn Signals into SLOs and Dashboards

Finally, promote a handful of metrics to service level objectives. A success-rate SLO with an error budget lets you alert on budget burn rather than on every transient blip — the same discipline that makes Fallback Routing for Failed Migration Jobs tractable, because a failure that is retried and recovered should not page anyone. Dashboards then plot the derived quantities: cost per gigabyte over time, throughput, drop rate, and error budget remaining.

#Production-Ready Implementation

The class below is a self-contained instrumentation wrapper for a conversion job. It binds a correlation ID, times the work, captures the resource signals, emits a structured terminal record, and returns a cost-annotated result on every exit path — including failures, which must never vanish silently.

# structlog>=24.1, pyarrow>=14.0
from __future__ import annotations

import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any, Callable

import structlog

log = structlog.get_logger()

UNIT_RATES = {
    "compute_usd_per_sec": 0.0000116,
    "storage_usd_per_gb_month": 0.023,
    "egress_usd_per_gb": 0.09,
}


@dataclass
class JobResult:
    job_id: str
    dataset_id: str
    rows_converted: int
    geometries_dropped: int
    source_bytes: int
    output_bytes: int
    compute_seconds: float
    usd_cost: float
    status: str


class InstrumentedConversion:
    """Wrap a single conversion in cost and observability instrumentation."""

    def __init__(self, dataset_id: str, source_bytes: int) -> None:
        self.dataset_id = dataset_id
        self.source_bytes = source_bytes
        self.job_id = uuid.uuid4().hex
        self._log = log.bind(job_id=self.job_id, dataset_id=dataset_id)

    def _cost(self, output_bytes: int, seconds: float) -> float:
        compute = seconds * UNIT_RATES["compute_usd_per_sec"]
        storage = (output_bytes / 1024**3) * UNIT_RATES["storage_usd_per_gb_month"]
        return round(compute + storage, 6)

    def run(self, work: Callable[[], dict[str, Any]]) -> JobResult:
        """Execute `work`, which must return rows/dropped/output_bytes.

        Emits a terminal structured record and returns a cost-annotated
        JobResult on both success and failure paths.
        """
        self._log.info("conversion.start", source_bytes=self.source_bytes)
        start = time.perf_counter()
        status = "ok"
        out: dict[str, Any] = {"rows": 0, "dropped": 0, "output_bytes": 0}
        try:
            out = work()
            if out.get("dropped", 0) > 0:
                status = "partial"
        except Exception as exc:  # noqa: BLE001 — terminal record must always emit
            status = "failed"
            self._log.error("conversion.error", error=str(exc), exc_type=type(exc).__name__)
        finally:
            seconds = round(time.perf_counter() - start, 4)
            result = JobResult(
                job_id=self.job_id,
                dataset_id=self.dataset_id,
                rows_converted=int(out.get("rows", 0)),
                geometries_dropped=int(out.get("dropped", 0)),
                source_bytes=self.source_bytes,
                output_bytes=int(out.get("output_bytes", 0)),
                compute_seconds=seconds,
                usd_cost=self._cost(int(out.get("output_bytes", 0)), seconds),
                status=status,
            )
            self._log.info("conversion.result", **asdict(result))
        return result


if __name__ == "__main__":
    def fake_work() -> dict[str, Any]:
        time.sleep(0.05)
        return {"rows": 812_004, "dropped": 4, "output_bytes": 210_000_000}

    job = InstrumentedConversion(dataset_id="parcels_2026", source_bytes=640_000_000)
    print(asdict(job.run(fake_work)))

The result object is deliberately flat and JSON-serialisable: it can be logged, pushed as metric dimensions, or appended to a cost ledger without transformation. Because the terminal record emits in the finally block, a job that raises still produces a status="failed" record with a measured duration and cost, so failed work is visible on dashboards rather than showing up as a gap.

#Metrics Reference

The signals below form a practical baseline for a conversion service. Emit each as the indicated instrument type; the Primary Use Case column shows the question each one answers.

#Failure Modes and Gotchas

#Frequently Asked Questions

What is the single most useful cost metric for a conversion pipeline?

Dollars per gigabyte of source data converted. It normalises across dataset sizes and lets you compare a small parcel Shapefile against a national road network on the same axis. Track it as a rolling trend rather than a single number: a sudden rise usually means a codec change, a retry storm, or a shift toward smaller files where fixed per-job overhead dominates.

Should I emit metrics from inside the conversion worker or scrape them afterwards?

Emit them from inside the worker as the job runs. Batch conversion jobs are short-lived and often ephemeral, so a scrape-based model like a Prometheus pull can miss workers that start and exit between scrape intervals. Push metrics through an OpenTelemetry exporter or a push gateway, and always emit a terminal record on both success and failure so no job disappears silently.

How do I keep observability overhead from inflating conversion cost?

Sample traces rather than recording every span, aggregate metrics in-process before export, and log at INFO for lifecycle events while reserving DEBUG for opt-in troubleshooting. For a well-tuned pipeline the telemetry path should add under 3% to wall-clock time and a negligible fraction of the compute bill. If it costs more, you are almost certainly logging per-row instead of per-job.

What SLOs make sense for a batch conversion service?

Three cover most needs: a success-rate objective (for example 99.5% of jobs complete without landing in the dead-letter path), a freshness objective (95% of datasets converted within their scheduled window), and a cost guardrail (dollars per gigabyte stays under a budgeted ceiling). Attach an error budget to the first two and alert on burn rate rather than on individual failures.

#Related

• Cost-per-GB Tracking for Conversion Pipelines
• Structured Logging and Metrics for Migration Jobs
• Building Batch Conversion Pipelines with Python
• Fallback Routing for Failed Migration Jobs
• ZSTD Compression Levels for Geospatial Data

← Back to Data Conversion & Migration Pipelines