Writing Idempotent Airflow DAGs for EDC Sync

An Airflow-orchestrated EDC sync runs clean for weeks, then a single task retry — or a well-meaning backfill — doubles the row count for one site, leaves a half-loaded visit after a mid-task crash, and produces a reconciliation gap no one can explain to an auditor. That symptom is what this deep-dive resolves: non-idempotent DAGs where a re-run does not converge to the same state. Clinical data managers, biotech and pharma Python engineers, and regulatory reviewers hit it because Airflow will re-execute tasks — retries, manual clears, scheduler failover, and backfills are all normal — and any task that appends instead of upserting, or reads “now” instead of its assigned window, turns each re-execution into a data-integrity defect. This page is the idempotency primitive inside Orchestrating EDC Sync Pipelines with Apache Airflow, which sits within the broader Automated EDC Ingestion & Sync Pipelines discipline. Built correctly, every task instance becomes safe to re-run at any time, which is the property that keeps the pipeline defensible under 21 CFR Part 11.

Idempotent Run Flow at a Glance #

Each run is pinned to a deterministic interval window; the extract reads only that window, the load upserts on natural keys, and the cursor is committed externally only after a verified write — so a retry re-runs the identical window with no duplication.

Root Cause: Why EDC DAGs Duplicate and Partially Load #

Idempotency failures in an EDC DAG almost always trace to one of three specific mistakes, and each maps to a normal Airflow behavior that a non-idempotent task turns into a defect.

The first is a task that reads a moving target instead of its assigned window. Airflow hands every run a data_interval_start and data_interval_end, but a task that filters the EDC extract on datetime.now() or a “last N hours” relative window will read different data every time it runs. When the scheduler retries that task, or a backfill replays it, the second execution pulls a different set of records than the first, so counts drift and the run is not reproducible.

The second is a mutable cursor stored in XCom. XCom is convenient, but it is scoped to a task instance and re-computed on retry, so using it to hold “the last timestamp I processed” means a retried task can advance the cursor twice or read a stale value. Worse, a cursor in XCom lives in the metadata database, and if it ever carries a subject-level value it becomes a PHI leak — the boundary the parent guide draws around Orchestrating EDC Sync Pipelines with Apache Airflow.

The third is a catchup flood. A DAG unpaused with catchup=True and a historical start_date enqueues one run per missed interval at once; if the load task appends rather than upserts, those overlapping runs race and double-load. The fix is not one trick but a discipline: pin each run to a deterministic window, make every write converge, and keep cursor state outside XCom.

Step-by-Step Implementation #

Each step owns one responsibility and produces a runnable building block. Compose them into a single sync DAG per study.

1. Scope each run to a deterministic data_interval window #

Derive the extract bounds strictly from the run’s interval, never from wall-clock time, so a retry or backfill of the same logical run always reads the identical window.

# 21 CFR Part 11 relevance: reproducibility — the extract window is a pure
# function of the run's data_interval, so any re-run reads the same records.
from airflow.decorators import task
from mypipeline import edc, store

@task(retries=3)
def extract(site: str, data_interval_start=None, data_interval_end=None) -> dict:
    # Bounds come ONLY from the interval — never datetime.now() or "last 4h".
    ref = edc.extract_site(
        study="ONC2024", site=site,
        since=data_interval_start,          # inclusive lower bound
        until=data_interval_end,            # exclusive upper bound
    )
    blob = ref.bytes
    uri = store.write(blob, key=f"ONC2024/{site}/{data_interval_start.isoformat()}")
    return {"site": site, "uri": uri, "sha256": store.sha256(blob),
            "records": ref.count}           # references only — no PHI in XCom

2. Upsert on natural keys so a re-run converges #

Make the load a no-op when the same records are seen twice by keying every row on its ODM natural-key tuple. A replayed run updates in place instead of inserting, so duplication is structurally impossible.

# ALCOA+ requirement: Consistent + Complete — the upsert converges, so a retried
# or backfilled load produces identical state with zero duplicate rows.
NATURAL_KEY = ("StudyOID", "SiteRef", "SubjectKey",
               "EventOID", "FormOID", "ItemGroupOID", "ItemOID")

def upsert_rows(conn, rows: list[dict]) -> int:
    conn.executemany(
        "INSERT INTO item_data "
        "(study, site, subject, event, form, item_group, item, value) "
        "VALUES (:StudyOID,:SiteRef,:SubjectKey,:EventOID,"
        ":FormOID,:ItemGroupOID,:ItemOID,:Value) "
        "ON CONFLICT(study, site, subject, event, form, item_group, item) "
        "DO UPDATE SET value = excluded.value",   # idempotent on the natural key
        rows,
    )
    conn.commit()
    return len(rows)

3. Externalize the cursor instead of storing it in XCom #

Persist the watermark in a durable store keyed by study and site, and commit it only after the upsert succeeds — so a crash between load and commit re-runs the window rather than skipping it.

# 21 CFR Part 11 relevance: the cursor is durable control-plane state, not XCom;
# it advances only after a verified write, so no window is silently skipped.
import sqlite3

def commit_cursor(db: sqlite3.Connection, study: str, site: str,
                  interval_end: str, batch_sha256: str) -> None:
    db.execute(
        "INSERT INTO edc_cursor(study, site, watermark, batch_sha256, ts) "
        "VALUES(?,?,?,?,strftime('%s','now')) "
        "ON CONFLICT(study, site) DO UPDATE SET "
        "watermark=excluded.watermark, batch_sha256=excluded.batch_sha256, "
        "ts=excluded.ts",
        (study, site, interval_end, batch_sha256),
    )
    db.commit()                                   # atomic: cursor advances or it does not

4. Guard the load with the interval bounds and a hash check #

Verify the payload hash and confine every write to the run’s interval before committing, so a stale or partial blob cannot corrupt state.

# ALCOA+ requirement: Accurate + Original — the write is gated on a hash match
# and confined to the run's window before the cursor is allowed to advance.
from airflow.exceptions import AirflowException
from mypipeline import store, warehouse, transform as xf

@task(retries=3)
def load(ref: dict, data_interval_end=None) -> dict:
    blob = store.read(ref["uri"])
    if store.sha256(blob) != ref["sha256"]:       # integrity gate
        raise AirflowException(f"hash mismatch for {ref['site']}")
    rows = xf.flatten_odm(blob)
    with warehouse.transaction() as conn:         # single atomic unit of work
        written = upsert_rows(conn, rows)
        commit_cursor(conn, "ONC2024", ref["site"],
                      data_interval_end.isoformat(), ref["sha256"])
    return {"site": ref["site"], "rows": written}

5. Fan out per site with independent idempotency and disarm catchup #

Map one load instance per site so each site’s window commits independently, and ship the DAG with catchup=False and max_active_runs=1 so no backfill flood or overlapping run can race the upsert.

# 21 CFR Part 11 relevance: controlled scheduling — no catchup storm, one active
# run per study, and per-site mapping isolates each site's idempotent window.
import pendulum
from airflow.decorators import dag

@dag(
    dag_id="edc_sync_study_ONC2024",
    schedule="0 */4 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,                                # no historical storm on deploy
    max_active_runs=1,                            # no overlapping runs of one study
    default_args={"retries": 3, "owner": "clinical-data-eng"},
)
def edc_sync():
    from mypipeline import config
    refs = extract.expand(site=config.active_sites(study="ONC2024"))
    load.expand(ref=refs)                         # each site commits independently

edc_sync()

Verification and Audit Trail #

A DAG is GxP-relevant software, so “the re-run was fine” must be provable from the record, not asserted. Confirm idempotency by running the same logical interval twice and asserting the second run inserts zero new rows and reproduces the batch hash; capture the per-run evidence below to a write-once store so an inspector can reconstruct every execution. The boundaries of what this control plane may record follow Audit Trail Boundaries in EDC Systems.

To confirm the fix end to end, assert three properties against a mocked EDC endpoint: an extract scoped to a fixed interval returns the same sha256 on every call; a load re-run for the same (site, data_interval) reports rows_new == 0; and a simulated crash between upsert_rows and commit_cursor leaves the cursor un-advanced so the next run re-processes the window. Genuine discrepancies the load surfaces feed Automated Clinical Query Generation rather than blocking the run, and unrecoverable batches route to Dead-Letter Queues and Error Recovery.

Edge Cases and Vendor-Specific Gotchas #

Backfills need a reset, not a re-run. Re-triggering a DAG run that already partially loaded can double-count if any task in the chain is not yet idempotent. Run historical loads with airflow dags backfill --reset-dagruns under a tracked change request so the run state is cleared first, and confirm the upsert converges before backfilling more than one interval.

Late-arriving EDC edits break a naive watermark. A source that back-dates an edit into an already-processed interval will be missed by a strictly forward-only cursor. Pair the interval scope here with an overlap window or change-data-capture watermark — the pattern in Implementing Watermark Cursors for EDC Delta Sync — so a re-opened form is re-pulled and upserted rather than lost.

Dynamic mapping over a shifting site list. If active_sites() returns a different list on a retry than on the original run, mapped task indices no longer line up and a cleared run can load the wrong site. Snapshot the site list into the run (via a params value or a first task whose output is the mapping source) so the fan-out is stable across every re-execution of that run.

Frequently Asked Questions #

Related #

• Orchestrating EDC Sync Pipelines with Apache Airflow — the parent guide this idempotency primitive plugs into.
• Implementing Watermark Cursors for EDC Delta Sync — externalized cursor design for late-arriving EDC edits.
• Dead-Letter Queues and Error Recovery — where a batch goes when retries cannot recover it.
• Automating Medidata Rave Data Pulls with Python — the natural-key upsert and audit-hash pattern this DAG orchestrates.
• Automated EDC Ingestion & Sync Pipelines — the parent reference for this discipline.