Orchestrating Annotation Pipelines with Airflow

An annotation pipeline that works when a person runs four scripts in order stops working the first night nobody runs them. The steps are not hard — harvest yesterday’s completed annotations, validate the geometry, export a training artifact, version it — but they need a schedule, credentials, retries that distinguish a timeout from a rejection, and a backfill story for the week the platform was down. That is the job Airflow does, and doing it badly produces the specific failure this topic exists to prevent: a DAG that, when rerun, exports today’s data into yesterday’s partition and silently doubles a class.

The design rule that makes everything else work is that each task reads only from its own data interval and writes to a path derived from it. Get that right and reruns are free, backfills reconstruct history correctly, and a failed task can be cleared without anyone reasoning about what it already did.

# Prerequisites & Toolchain Alignment

pip install "apache-airflow==2.9.3" \
            "apache-airflow-providers-amazon==8.25.0" \
            geopandas==0.14.4 shapely==2.0.6 pyproj==3.6.1 dvc[s3]==3.51.2

Three assumptions the DAG below makes:

• The annotation platform can be queried by a closed time window. Label Studio, CVAT and most hosted platforms support filtering by update timestamp and review state. If yours only offers “changed since my last poll”, wrap it in a store that records what each interval saw, because that endpoint cannot be backfilled.
• The validators already exist as a library. Airflow should call the same checks the CI/CD gate runs on pull requests, not a second copy that drifts from it.
• Dataset versioning is content-addressed. The final task pushes a manifest; if the content is unchanged the version must not advance, which is the property tracking annotation changes with SHA hashing provides.

# Building the DAG

# Step 1 — Harvest a Closed Window

The single most important line in the DAG is the query bound. data_interval_start and data_interval_end are supplied by Airflow for the interval being processed, not for now, which is what makes a backfill mean anything.

from __future__ import annotations
import json
from datetime import datetime, timedelta
from pathlib import Path

import httpx
from airflow.decorators import dag, task

PLATFORM = "https://labels.internal/api"

@task(retries=3, retry_delay=timedelta(minutes=2), retry_exponential_backoff=True)
def harvest(data_interval_start: datetime, data_interval_end: datetime, ds: str) -> str:
    """Pull annotations REVIEWED within this interval; write one file keyed on it."""
    out = Path(f"/data/raw/{ds}/features.geojson")
    out.parent.mkdir(parents=True, exist_ok=True)
    params = {
        "reviewed_after": data_interval_start.isoformat(),
        "reviewed_before": data_interval_end.isoformat(),
        "state": "approved",          # never work in progress
        "page_size": 500,
    }
    features: list[dict] = []
    with httpx.Client(timeout=60.0) as client:
        url = f"{PLATFORM}/annotations"
        while url:
            resp = client.get(url, params=params)
            if 400 <= resp.status_code < 500 and resp.status_code != 429:
                raise RuntimeError(f"platform rejected the query: {resp.status_code} {resp.text[:200]}")
            resp.raise_for_status()
            body = resp.json()
            features.extend(body["results"])
            url, params = body.get("next"), None
    out.write_text(json.dumps(
        {"type": "FeatureCollection", "features": features},
        indent=2, sort_keys=True) + "\n")
    return str(out)

The state filter is not a detail. Harvesting in-progress annotations means a tile half-drawn at midnight enters the dataset, gets exported, and is then re-harvested the next night in its finished form — two versions of one feature, both real, with nothing to reconcile them.

The explicit 4xx check before raise_for_status distinguishes the two failure kinds: a malformed query should stop the DAG immediately, while a 429 or a 502 falls through to Airflow’s retry policy.

# Step 2 — Validate as Its Own Task

Validation is a separate task so that its failure is legible in the DAG graph and so a fix can be applied and only that task cleared.

@task
def validate(raw_path: str, ds: str) -> str:
    """Run the same checks the pull-request gate runs. Fail the task on any error."""
    import geopandas as gpd
    from annotation_gates import check_geometry, check_crs, check_schema   # shared library

    gdf = gpd.read_file(raw_path)
    errors: list[str] = []
    errors += check_geometry(gdf)
    errors += check_crs(gdf, expected="EPSG:4326")
    errors += check_schema(gdf, schema_path="/config/label_schema.json")

    report = Path(f"/data/reports/{ds}/validation.json")
    report.parent.mkdir(parents=True, exist_ok=True)
    report.write_text(json.dumps({"errors": errors, "n_features": len(gdf)}, indent=2) + "\n")
    if errors:
        raise ValueError(f"{len(errors)} validation error(s); first: {errors[0]}")
    return raw_path

Writing the report before raising is deliberate: the task fails, the DAG stops, and the artifact explaining why is already on disk rather than trapped in a log the on-call engineer has to page through.

# Step 3 — Export, Then Version Only If Something Changed

@task
def export(validated_path: str, ds: str) -> str:
    import geopandas as gpd
    gdf = gpd.read_file(validated_path).to_crs("EPSG:4326")
    out = Path(f"/data/export/{ds}/labels.parquet")
    out.parent.mkdir(parents=True, exist_ok=True)
    gdf.to_parquet(out, index=False, compression="zstd")
    return str(out)

@task
def version(export_path: str, ds: str) -> str | None:
    """Hash the export; push a new dataset version only when the content moved."""
    import hashlib
    import subprocess

    digest = hashlib.sha256(Path(export_path).read_bytes()).hexdigest()
    marker = Path("/data/manifests/last_digest.txt")
    if marker.exists() and marker.read_text().strip() == digest:
        print(f"{ds}: content unchanged ({digest[:12]}) — no new version")
        return None
    subprocess.run(["dvc", "add", export_path], check=True)
    subprocess.run(["dvc", "push"], check=True)
    marker.parent.mkdir(parents=True, exist_ok=True)
    marker.write_text(digest + "\n")
    return digest

A nightly DAG that tags a version every night produces 365 versions a year and no way to find the ones that mattered. The digest comparison is what keeps the history readable.

# Step 4 — Wire It Up

@dag(
    dag_id="annotation_harvest",
    schedule="0 2 * * *",                 # 02:00, after the review shift closes
    start_date=datetime(2026, 1, 1),
    catchup=False,                        # turn on deliberately for a backfill
    max_active_runs=1,                    # the export path is per interval, not per run
    default_args={"owner": "annotation-platform", "retries": 0},
    tags=["annotation", "geospatial"],
)
def annotation_harvest():
    raw = harvest()
    ok = validate(raw)
    art = export(ok)
    version(art)

annotation_harvest()

max_active_runs=1 matters more than it looks. Two concurrent runs of different intervals write different paths and are safe; two concurrent runs of the same interval — which a manual trigger during a scheduled run produces — race on one file.

# Pipeline Parameters & Configuration Reference

# Edge Cases & Gotchas

A backfill that hammers the annotation platform. Fifty parallel interval runs each paginating the API is indistinguishable from an attack. Put the platform tasks in a pool of two, so a backfill is slow rather than disruptive.

Timezones on the interval bounds. Airflow’s intervals are timezone-aware; annotation platforms frequently return naive local timestamps. Comparing the two silently drops or duplicates an hour twice a year. Normalise to UTC at the boundary and store UTC in the harvested file.

Reviewers who reopen an approved annotation. A feature approved on Monday and corrected on Wednesday appears in two intervals. That is correct and the versioning layer handles it — the later export supersedes the earlier — but only if the export writes whole batches rather than appending.

Secrets in the DAG file. The platform token belongs in a connection or a secrets backend, not in the DAG. DAG files are parsed constantly and their contents end up in logs and in the UI’s code view.

A validation failure that blocks the schedule forever. If today’s harvest fails validation and the DAG has depends_on_past=True, every subsequent night fails too. Keep intervals independent and let the alert, not the scheduler, get the problem fixed.

# Integration & Automation Hooks

Triggering training. The version task returns a digest or None. Feed that into a downstream DAG trigger so retraining starts only on a real change — the same condition triggering retraining from new annotations with DVC applies from the DVC side.

Sharing the validators with CI. Import the same annotation_gates package the pull-request gate uses. Two copies of the rules is the reliable way to get a batch that passes CI and fails the DAG.

Reporting to the annotation team. A short Slack or email summary — features harvested, validation errors, whether a version was created — closes the loop for the people whose work the DAG is consuming. Silence on success and noise on failure trains everyone to ignore it; a one-line daily summary does not.

# Validation & Testing

DAGs deserve tests that do not need a scheduler.

from airflow.models import DagBag

def test_dag_imports_without_errors() -> None:
    bag = DagBag(include_examples=False)
    assert bag.import_errors == {}, bag.import_errors

def test_paths_are_interval_scoped() -> None:
    """Every artifact path must contain the run's date, or reruns overwrite the wrong thing."""
    import inspect
    from dags.annotation_harvest import harvest, validate, export
    for fn in (harvest, validate, export):
        src = inspect.getsource(fn.function)
        assert "{ds}" in src, f"{fn.function.__name__} writes a path with no interval in it"

def test_client_error_is_not_retried(monkeypatch) -> None:
    """A 422 must raise immediately rather than falling into the retry policy."""
    import httpx, pytest
    from dags.annotation_harvest import harvest

    class Rejecting(httpx.Client):
        def get(self, *a, **k):
            return httpx.Response(422, text="bad filter", request=httpx.Request("GET", "http://x"))

    monkeypatch.setattr(httpx, "Client", Rejecting)
    with pytest.raises(RuntimeError, match="platform rejected"):
        harvest.function(datetime(2026, 7, 1), datetime(2026, 7, 2), "2026-07-01")

The second test is the one worth keeping. It feeds the DAG’s own source to an assertion that would reject the most damaging mistake in this whole topic — an artifact path with no interval in it — rather than trusting that nobody will ever write one.

# Frequently Asked Questions

# The Property Worth Protecting

Everything in this topic is downstream of a single property: a task reads only from its own data interval and writes only to a path derived from it. It is worth restating because it is the property that erodes first. A quick fix that writes to a shared “latest” path, a convenience symlink, an export step that appends rather than replaces — each is individually reasonable and each removes the guarantee that makes reruns free and backfills correct. The test in the validation section exists because that erosion is invisible in review and expensive to discover.

# Related

• CI/CD Gates for Annotation Datasets — the validators this DAG calls, and why both paths must share one implementation
• Implementing DVC for Geospatial Training Data — the data-dependency half of the pipeline that the version task delegates to
• Human-in-the-Loop Validation Cycles — the review states the harvest filter depends on
• Serving Imagery Tiles to Annotation Tools — the service side of the same toolchain, deliberately outside the DAG

Orchestration is the connective tissue of the broader Labeling Workflows & Toolchain Integration pipeline, turning a sequence of scripts into something that runs unattended.