Geospatial Orchestration Hub

How to Chain GDAL Tasks in Prefect

Chaining GDAL utilities in Prefect means one task per invocation, with each task returning the path it produced and taking its predecessor’s path as an argument. That argument is what creates the dependency edge — Prefect infers the graph from data flow, so a task that ignores its predecessor’s return value runs concurrently with it no matter what order you wrote the calls in. Wrap each subprocess so a non-zero exit becomes a Python exception carrying GDAL’s own stderr, and the flow ends up both correct and debuggable.

When to Use This Pattern

  • The pipeline is a sequence of GDAL command-line steps — ogr2ogr, gdalwarp, gdal_translate, gdalbuildvrt — which is the normal shape for vector and raster preparation.
  • Steps take long enough to be worth retrying individually, roughly anything over about thirty seconds.
  • You want per-step visibility in the Prefect UI: which step failed, how long each took, what GDAL actually said.
  • The same steps run over many partitions, so the chain will later be mapped over a list of municipalities or tiles.

If the whole job is one ogr2ogr call, do not build a chain. A single task with retries is the right amount of machinery, and adding a graph around one node costs clarity for nothing.

Complete Working Example

The wrapper below is the only piece of infrastructure needed. Everything after it is ordinary Prefect.

from __future__ import annotations

import subprocess
from pathlib import Path
from typing import Sequence

from prefect import flow, task, get_run_logger


class GdalError(RuntimeError):
    """A GDAL utility exited non-zero. Carries the tail of its stderr."""


def run_gdal(argv: Sequence[str], timeout: float = 1800.0) -> None:
    """Invoke a GDAL utility and turn its exit code into a Python exception."""
    logger = get_run_logger()
    logger.info("running %s", " ".join(argv[:4]))
    proc = subprocess.run(
        argv,
        capture_output=True,
        timeout=timeout,
        check=False,
        start_new_session=True,     # so a cancellation can reach the whole group
    )
    if proc.returncode != 0:
        # GDAL writes the useful part of the diagnosis to stderr; without this the
        # Prefect UI shows "CalledProcessError: returned 1" and nothing else.
        raise GdalError(proc.stderr.decode("utf-8", "replace").strip()[-800:])


@task(retries=2, retry_delay_seconds=[30, 120], timeout_seconds=1800)
def extract_layer(source: str, out_dir: Path, layer: str) -> Path:
    """WFS or file source into a local GeoPackage."""
    out = out_dir / "01_extract.gpkg"
    run_gdal([
        "ogr2ogr",
        "-f", "GPKG", str(out), source,
        "-nln", layer,
        "-lco", "SPATIAL_INDEX=YES",
        "-skipfailures",              # keep the batch; failures land in the log
    ])
    return out


@task(retries=1, timeout_seconds=3600)
def reproject_layer(src: Path, out_dir: Path, target_srs: str = "EPSG:25832") -> Path:
    out = out_dir / "02_projected.gpkg"
    run_gdal([
        "ogr2ogr",
        "-f", "GPKG", str(out), str(src),
        "-t_srs", target_srs,
        # Explicit geometry type: make_valid and repair steps can change it, and a
        # PostGIS typed column will reject the surprise later rather than here.
        "-nlt", "PROMOTE_TO_MULTI",
        "-lco", "SPATIAL_INDEX=YES",
    ])
    return out


@task(retries=1, timeout_seconds=3600)
def load_postgis(src: Path, dsn: str, table: str) -> str:
    run_gdal([
        "ogr2ogr",
        "-f", "PostgreSQL", f"PG:{dsn}", str(src),
        "-nln", table,
        "-lco", "GEOMETRY_NAME=geom",
        "-lco", "FID=gid",
        "-lco", "SPATIAL_INDEX=GIST",
        "-overwrite",
        # One transaction for the whole load: a failure leaves no partial table.
        "--config", "PG_USE_COPY", "YES",
    ])
    return table


@flow(name="vector-prepare", log_prints=True)
def vector_prepare(source: str, scratch: Path, dsn: str) -> str:
    scratch.mkdir(parents=True, exist_ok=True)
    extracted = extract_layer(source, scratch, layer="features")
    # Passing `extracted` — not just calling the tasks in order — is what makes
    # Prefect wait. A task that ignores its predecessor's return runs in parallel.
    projected = reproject_layer(extracted, scratch)
    return load_postgis(projected, dsn, table="features_staging")
Data flow is what creates the edgeWhen each task takes its predecessor's return value, Prefect runs them in sequence. When each task uses a hard-coded path instead, Prefect sees no dependency and runs all three concurrently.PATHS PASSED — a real chainextract → Pathreproject → PathloadFIXED PATHS — three concurrent tasks racing one fileextractreprojectloadall at onceThe lower flow often “works” on small test data, because extract finishes before reproject opens the file.
The failure in the lower case is intermittent and scales with data size — the worst combination, and the reason to make every handoff an argument.

Three details in the tasks above are load-bearing. start_new_session=True costs nothing here and is what lets a cancellation later reach GDAL’s whole process group rather than just the shell. Returning a Path rather than None is what creates the graph edge, so it is not a stylistic choice. And the retry counts differ per step deliberately: extraction talks to a network and deserves two retries, while the load talks to a database whose failures are usually schema mismatches that will not resolve themselves, so one is plenty.

Retry settings follow what each step talks toThe extract step talks to a network and gets two retries with backoff. Reprojection is local and gets one. The load talks to a database and gets one retry with a longer timeout.steptalks toretrieswhyextracta WFS endpoint2 + backofftransient by naturereprojectthe local disk1only OOM is transientloadPostGIS1schema errors repeat
A uniform retry count across a chain is a sign nobody asked what each step can actually fail on. Two of these three failures do not improve with another attempt.

Parameter & Option Reference

Option Where Purpose Spatial notes
-nlt PROMOTE_TO_MULTI ogr2ogr Normalise geometry type Repair steps turn polygons into multipolygons; promoting everything removes the whole class of type mismatch.
-lco SPATIAL_INDEX=YES GeoPackage Build a spatial index Costs seconds on write and saves minutes on every subsequent read.
--config PG_USE_COPY YES PostgreSQL Bulk load via COPY Several times faster than per-row INSERT for large loads.
-skipfailures ogr2ogr Continue past bad features Use only where the failures are captured elsewhere; silently dropping features is worse than failing.
-t_srs ogr2ogr Target CRS Always explicit. Relying on the source’s declaration is the most common cause of misplaced data.
timeout_seconds Prefect task Backstop Above the subprocess timeout, so your own error message wins.
retries Prefect task Per-step retry Only on steps that are idempotent — which, for a fixed output path, they are.

Verification & Testing

Test the wrapper, the graph shape, and the failure message — in that order of value.

import pytest


def test_gdal_error_carries_stderr(tmp_path) -> None:
    with pytest.raises(GdalError) as excinfo:
        run_gdal(["ogr2ogr", "-f", "GPKG", str(tmp_path / "x.gpkg"), "/nonexistent.shp"])
    # The message must name the file, not just the exit code.
    assert "nonexistent" in str(excinfo.value).lower()


def test_flow_runs_steps_in_order(tmp_path, monkeypatch) -> None:
    calls: list[str] = []
    monkeypatch.setattr(
        "mymodule.run_gdal", lambda argv, **kw: calls.append(argv[0] + ":" + argv[3])
    )
    vector_prepare("fixtures/sample.gpkg", tmp_path, dsn="dbname=test")
    # Each step must have consumed the previous step's output path.
    assert "01_extract.gpkg" in calls[1]
    assert "02_projected.gpkg" in calls[2]

Against real data, the check that matters is on the output rather than on the flow. ogrinfo answers both questions — did every feature arrive, and did it arrive in the right CRS — in one command:

# Feature count and SRS of what landed, compared with what was extracted.
ogrinfo -so -al scratch/01_extract.gpkg   | grep -E "Feature Count|PROJCRS|GEOGCRS"
ogrinfo -so -al scratch/02_projected.gpkg | grep -E "Feature Count|PROJCRS"

# And in the database, the same two facts:
psql "$DSN" -c "SELECT count(*), ST_SRID(geom) FROM features_staging GROUP BY 2;"
What a failure looks like in the runExtract and reproject completed. Load failed with GDAL's stderr attached. The downstream task is not attempted, and the two completed intermediates remain on disk.extractCompleted · 4m12sreprojectCompleted · 7m03sloadFailed · attempt 2/2not attemptedGdalError: ERROR 1: Geometry type of `Polygon` is not compatible withthe geometry type of layer `features_staging` (MultiPolygon)02_projected.gpkg still on disk — the re-run starts at load, not at extract
The error text is GDAL’s own, which is what makes it actionable. Without the wrapper this reads “CalledProcessError: returned non-zero exit status 1”.

There is one more habit worth building around this: keep the exact ogr2ogr command line in the log, not a summary of it. The wrapper above logs only the first four arguments to keep the log readable, which is a reasonable default for a healthy run — but on failure, the full argument vector is what lets someone paste the command into a terminal and reproduce the problem in ten seconds. Attaching it to the exception rather than to the info log gives you both: quiet when things work, complete when they do not.

Common Pitfalls

  • Calling tasks in order without passing values. Prefect builds the graph from data flow. Three tasks that each hard-code their paths have no edges between them and will run concurrently — usually fine on small test data and reliably broken in production.
  • Swallowing GDAL’s stderr. subprocess.run(check=True) raises with the exit code and discards the message that says which geometry, which field, which file. Capture stderr and put its tail in the exception.
  • Retrying a non-idempotent step. ogr2ogr with -append into PostGIS duplicates on every retry. Use -overwrite for staging tables, and an idempotent upsert for the real one.
  • Using -skipfailures as a default. It converts a hard failure into a silently smaller dataset. Where the source genuinely contains junk, pair it with a feature-count assertion so the drop is visible, or route the bad features to a dead-letter queue.
  • Letting scratch paths collide between concurrent runs. Two flow runs writing 01_extract.gpkg in the same directory will corrupt each other in ways that look like GDAL bugs. Namespace scratch by flow-run id.

Frequently Asked Questions

Should each GDAL call be its own task, or should one task do several?

One task per invocation, as long as the invocations are individually slow. The granularity buys per-step retries, per-step timing in the UI, and a re-run that starts where it failed. The exception is a handful of fast calls — building a VRT, reading metadata — where the orchestration overhead exceeds the work; group those.

How do I map this chain over many municipalities?

.map() over the list, with the scratch directory parameterised per item so the intermediates do not collide. That turns a chain into a fan-out, and the fan-out needs a concurrency limit or it will start four hundred gdalwarp processes — see limiting DAG fan-out with concurrency groups.

Is the GDAL Python API better than the command line here?

For chaining, the command line is usually easier to operate: it is a separate process, so it can be killed cleanly, its memory is bounded by the process rather than by your worker, and its arguments are copy-pasteable into a terminal when you are debugging. The Python API wins when you need to inspect data between operations without a round trip through a file.

What should the task's timeout be relative to the subprocess timeout?

The Prefect timeout should be the larger of the two, so that in normal operation your own subprocess.run(timeout=…) fires first and raises a message you wrote. The orchestrator’s limit is the backstop for the case where your own bookkeeping is wrong, which is exactly when a generic message is acceptable. Timeout budgets and cancellation covers how to derive both.

← Building ETL Chains for Vector Data

Related in this section