# Automating CRS Reprojection in Airflow

When domains ingest data in whatever CRS their source vendor emits — a cadastral extract in EPSG:32633, a survey feed in a local grid, a partner drop in web mercator — the mesh cannot federate until every product lands in one canonical reference system. Automating CRS reprojection in Airflow builds a scheduled DAG that reprojects each incoming dataset to EPSG:4326 with ogr2ogr and PostGIS ST_Transform, writes to a deterministic idempotent output path, and gates publication behind a data-quality check so a retried run never produces a second, divergent artifact. It is the Airflow implementation of the patterns in Orchestrating Spatial Pipelines in Python, within Spatial Pipeline Orchestration & Observability; the Prefect counterpart is Building Idempotent Spatial DAGs with Prefect.

# Prerequisites

Figure — An Airflow DAG: detect source CRS, reproject to EPSG:4326 on a deterministic path, run a data-quality check, then publish only on pass.

# Step-by-Step Implementation

Each step is verifiable before you proceed. The invariant is that the output path is a pure function of the layer and source CRS, so a retried task overwrites the same object rather than creating a new one.

# 1. Detect the source CRS deterministically

Read the source reference with ogrinfo and pin it into an XCom, so downstream tasks derive the output path from a known value rather than guessing.

# dag.py
from airflow.decorators import dag, task
from datetime import datetime
import subprocess, re, os

@dag(schedule="0 3 * * *", start_date=datetime(2026, 1, 1), catchup=False)
def crs_reprojection():

    @task
    def detect_crs(src: str) -> str:
        out = subprocess.check_output(["ogrinfo", "-so", "-al", src], text=True)
        m = re.search(r"ID\[\"EPSG\",(\d+)\]", out)
        return f"EPSG:{m.group(1)}"  # e.g. EPSG:32633

Verify: confirm the detected CRS matches the source.

ogrinfo -so -al "$SRC_URI" | grep -Eo 'EPSG",[0-9]+' | head -1

# 2. Reproject to EPSG:4326 on a deterministic path

A BashOperator-style task calls ogr2ogr with -t_srs EPSG:4326. The output path is derived from the layer and source CRS, so re-running overwrites rather than appends. The -overwrite flag makes the write idempotent.

    @task
    def reproject(src: str, src_crs: str, layer: str) -> str:
        out = f"{os.environ['REPROJECT_OUT']}/{layer}/{src_crs.replace(':','_')}__4326/part.gpkg"
        os.makedirs(os.path.dirname(out), exist_ok=True)
        subprocess.run([
            "ogr2ogr", "-overwrite",
            "-s_srs", src_crs, "-t_srs", "EPSG:4326",
            "-nln", layer, out, src,
        ], check=True)  # domain-scoped credential via GDAL config
        return out

Verify: confirm the output is genuinely in EPSG:4326.

ogrinfo -so -al "$REPROJECT_OUT/parcels/EPSG_32633__4326/part.gpkg" | grep -i '4326'

# 3. Run the data-quality check in PostGIS

Load the reprojected features and assert feature count and geometry validity with ST_IsValid. This is the gate: it must raise on any invalid geometry so the DAG stops before publication.

    @task
    def quality_check(path: str, layer: str) -> str:
        subprocess.run([
            "ogr2ogr", "-f", "PostgreSQL",
            os.environ["AIRFLOW_CONN_DOMAIN_PG"], path, "-nln", f"stg_{layer}",
        ], check=True)
        sql = f"SELECT count(*) FILTER (WHERE NOT ST_IsValid(geom)) FROM stg_{layer};"
        invalid = int(run_psql_scalar(sql))
        if invalid:
            raise ValueError(f"quality gate: {invalid} invalid geometries in {layer}")
        return path

Verify: count invalid geometries directly.

psql "$AIRFLOW_CONN_DOMAIN_PG" -tAc \
  "SELECT count(*) FILTER (WHERE NOT ST_IsValid(geom)) FROM stg_parcels;"

# 4. Publish only on a passing check and wire the dependency

Publication runs strictly after the quality gate, so a failed check leaves the previous product version untouched.

    @task
    def publish(path: str, layer: str, version: str):
        write_catalog_entry(layer, version=version, artifact=path, crs="EPSG:4326")

    src = os.environ["SRC_URI"]
    crs = detect_crs(src)
    reproj = reproject(src, crs, "parcels")
    checked = quality_check(reproj, "parcels")
    publish(checked, "parcels", "v1.2.0-crs:EPSG:4326-res:10m")

crs_reprojection()

Verify: trigger the DAG and confirm task ordering and success.

airflow dags test crs_reprojection 2026-07-14

# 5. Confirm idempotency across a re-run

Re-running over unchanged input must produce a byte-stable output path and no duplicate objects.

airflow tasks test crs_reprojection reproject 2026-07-14
ls "$REPROJECT_OUT/parcels/EPSG_32633__4326/" | wc -l   # stays 1 across runs

# Configuration Reference

# Common Failure Modes & Fixes

ogr2ogr reprojects but coordinates look wrong. Root cause: the source CRS was misdetected or omitted, so GDAL assumed the wrong -s_srs. Fix: pass the detected -s_srs explicitly from step 1 and re-verify with ogrinfo -so -al ... | grep 4326 plus a spot-check of a known coordinate.

A retry creates a second output file. Root cause: the output path included a timestamp or run id, so each attempt wrote a new object. Fix: derive the path only from layer and src_crs, add -overwrite, and confirm the directory holds exactly one file across runs.

Quality check passes but downstream sees invalid polygons. Root cause: the gate queried the wrong staging table or ran before the load completed. Fix: make quality_check depend on the load, target stg_{layer}, and assert the invalid count is zero with psql.

Axis order flips (lat/lon swapped). Root cause: a datum or authority-compliant axis-order difference between source and EPSG:4326. Fix: add -t_srs EPSG:4326 with GDAL OGR_CT_FORCE_TRADITIONAL_GIS_ORDER=YES when the source uses authority order, then re-verify a sample coordinate.

DAG reprojects the whole history every night. Root cause: no change detection, so unchanged sources are reprocessed. Fix: gate the reproject on a source content hash or modification time, skipping the task when the input is unchanged.

# FAQ

# Related

• Up to the parent: Orchestrating Spatial Pipelines in Python
• Building Idempotent Spatial DAGs with Prefect — the same idempotency goal via cache_key_fn
• Topology Validation with Dagster Asset Checks — declaring the validity gate as an asset check
• Spatial Pipeline Orchestration & Observability — the section overview and SLA baselines
• CRS Governance and Reprojection Standards — why EPSG:4326 is the canonical interchange CRS