Backfilling a Month of Annotation Harvests

The harvest DAG was broken for four weeks — a rotated credential nobody noticed — and the dataset has a month-shaped hole in it. Backfilling looks like one command, and on a DAG whose tasks are interval-scoped it very nearly is. On a DAG whose tasks are not, the same command writes today’s data into thirty historical partitions and doubles a class in every one. This guide covers the check that tells you which DAG you have, the pool that keeps the backfill from taking the annotation platform down with it, and the dry run that proves the whole thing on one interval first.

# Why This Matters in Geospatial Pipelines

A gap in an annotation dataset is not merely missing rows. Annotations are harvested by review date, and the tiles reviewed in a given week cluster geographically — that is how work is assigned. A four-week gap is therefore usually a spatial gap, and a model trained without it has a hole in its coverage rather than slightly less data. That is also why filling the gap with empty files is worse than leaving it: it converts a visible absence into an apparent measurement.

# Step-by-Step Implementation

# Step 1 — Prove the Paths Are Interval-Scoped

Before triggering anything, confirm the property the whole operation rests on. The cheapest check is on the code, not on the data.

import inspect

def assert_interval_scoped(*task_fns) -> None:
    """Every task must write to a path containing its own data interval."""
    for fn in task_fns:
        src = inspect.getsource(getattr(fn, "function", fn))
        writes = [ln for ln in src.splitlines() if "Path(" in ln or "open(" in ln]
        for ln in writes:
            if "{ds}" not in ln and "data_interval" not in ln:
                raise AssertionError(
                    f"{getattr(fn, '__name__', fn)}: writes a path with no interval in it:\n  {ln.strip()}")

If this fails, do not backfill. Fix the paths first, because a backfill over shared paths is not recoverable by re-running — the original contents are gone.

# Step 2 — Cap Concurrency With a Pool

Thirty parallel runs each paginating an annotation API is a denial-of-service attack on your own platform. A pool makes the backfill polite by construction.

airflow pools set annotation_api 2 "Serialises platform-facing tasks during backfills"

@task(pool="annotation_api", retries=3, retry_delay=timedelta(minutes=2),
      retry_exponential_backoff=True, execution_timeout=timedelta(minutes=30))
def harvest(data_interval_start, data_interval_end, ds: str) -> str:
    ...

Only the platform-facing task needs the pool. Validation, export and versioning are local and can run at whatever parallelism the workers allow, so the backfill is serialised exactly where it needs to be and nowhere else.

# Step 3 — Dry-Run One Interval and Compare

Pick an interval that ran successfully before the outage, re-run it into a scratch prefix, and compare the result against what it produced originally. If the two match, the DAG is genuinely idempotent and the range is safe.

import hashlib
from pathlib import Path

def compare_reharvest(original: str, reharvested: str) -> None:
    """A re-run of a past interval must reproduce that interval's content."""
    a = hashlib.sha256(Path(original).read_bytes()).hexdigest()
    b = hashlib.sha256(Path(reharvested).read_bytes()).hexdigest()
    if a != b:
        raise AssertionError(
            f"re-harvest of the same interval differs\n  was {a[:16]}\n  now {b[:16]}\n"
            "the query is not interval-bounded, or reviews were edited since")

A mismatch has two innocent explanations and one alarming one. Reviews genuinely edited since that day will change the content legitimately; so will a taxonomy change. A query bounded by “since last run” rather than by the interval will also differ, and that is the case that must stop the backfill.

# Step 4 — Run the Range, Oldest First

airflow dags backfill annotation_harvest \
  --start-date 2026-06-08 --end-date 2026-07-05 \
  --reset-dagruns --rerun-failed-tasks

Oldest first is the default and is the behaviour you want: downstream consumers see history arrive in order, and progress is legible as a gap closing from one end rather than as thirty runs in unpredictable states.

Watch the pool, not the DAG. With a pool of two and a four-minute harvest, twenty-eight intervals take roughly an hour, and a run queue that is not draining at that rate means the platform is throttling — which the retry policy will handle, more slowly.

# Step 5 — Reconcile the Version Count

A correct backfill produces one dataset version per genuinely distinct day, not one per run.

def reconcile(expected_days: int, versions_created: int, unchanged_days: int) -> None:
    if versions_created + unchanged_days != expected_days:
        raise AssertionError(
            f"{expected_days} intervals backfilled but {versions_created} versions "
            f"and {unchanged_days} no-change days recorded — the content check is not running")

If the count comes out at one version per interval including days with no reviews, the content-hash comparison in the version task is not being consulted, and the dataset history has just gained a month of noise. That check is the one described in triggering retraining from new annotations with DVC.

# Parameters and Thresholds Reference

# Common Errors and Fixes

Every backfilled day produces a new dataset version Root cause: the version task is not comparing content hashes. Fix: restore the digest comparison; then re-run the range, which is safe precisely because the paths are interval-scoped.

The platform starts returning 429 halfway through Root cause: the pool is larger than the platform’s rate allowance, or another job shares the token. Fix: reduce the pool to one and let the retry backoff absorb the rest. The backfill takes twice as long and nobody else notices it.

Backfilled files are all empty Root cause: the harvest filters on review state and those reviews were completed before the state field existed, or beyond the retention window. Fix: stop the backfill, record the gap in the dataset’s provenance, and do not ship empty files as though they were measurements.

The DAG re-runs but downstream training does not Root cause: the training trigger fires on a dataset version, and the backfill produced versions with historical timestamps that the trigger’s watermark has already passed. Fix: trigger training once, explicitly, after the backfill completes — it is one decision, not twenty-eight.

# Frequently Asked Questions

# Related

• Orchestrating Annotation Pipelines with Airflow — the DAG design whose interval-scoped paths make this operation safe
• Using DVC Pipelines for Automated Dataset Snapshots — the content-hash comparison that keeps a backfill from flooding the version history
• Human-in-the-Loop Validation Cycles — the review states the harvest query filters on

Backfilling is one operation within Orchestrating Annotation Pipelines with Airflow, part of Labeling Workflows & Toolchain Integration.