Orchestrating EDC Sync Pipelines with Apache Airflow

An Electronic Data Capture (EDC) sync program that starts as a single cron job rarely survives contact with a real multi-site study: extraction has to fan out per study and per site, wait for the source system to actually be ready, retry transient failures without duplicating loads, raise an alert when a safety-relevant feed is late, and leave a defensible record of every one of those decisions. Apache Airflow is the orchestration layer most clinical data teams reach for, and this guide — part of Automated EDC Ingestion & Sync Pipelines — covers how to build Airflow DAGs that are idempotent, auditable, and validatable rather than a tangle of unscheduled scripts. For clinical data managers, biotech and pharma Python engineers, and regulatory reviewers, the orchestrator is not neutral plumbing: it decides when data moves, who is attributable for a run, and whether a retried task corrupts the analysis dataset. The single most important rule appears before any code below, because getting it wrong is a reportable data-integrity event — never put protected health information (PHI) in XCom or the Airflow metadata database. XCom and the metadata DB pass cursors, keys, and counts; clinical values live in an external, access-controlled store. Everything else in this page builds on that boundary so that each scheduled run stays attributable, contemporaneous, and reconstructable under 21 CFR Part 11 and ICH E6(R3).

The Orchestrated Sync at a Glance #

A readiness sensor gates the run; a mapped extract task fans out one instance per site; validated batches flow through transform and load; and a terminal audit task records the run before the DAG is allowed to succeed.

Concepts and Prerequisites #

Airflow orchestrates when and in what order work runs; it is not itself the extraction or transformation logic. That separation matters for validation, because the DAG definition is a configuration artifact under change control while the task callables are the tested units. Before writing a DAG you should be fluent in Airflow’s execution model — the data_interval_start/data_interval_end window that defines each run, the difference between the logical logical_date (formerly execution_date) and wall-clock time, and how catchup interacts with schedule and start_date. You should also understand the CDISC ODM envelope the source EDC exposes (see CDISC ODM vs CDASH Schema Mapping) and the audit-trail expectations carried over from Audit Trail Boundaries in EDC Systems. The extraction that these DAGs schedule is documented in Python ETL for EDC Data Extraction; orchestration only decides how those extractors are invoked, retried, and audited.

The reference implementation assumes a pinned, version-controlled dependency set so validated behavior is reproducible across IQ/OQ/PQ environments:

Two environment assumptions are non-negotiable. First, the DAG process and its workers never see raw credentials in code — Connections and secrets resolve through a configured secrets backend at runtime. Second, the metadata database is treated as a control-plane store, not a data store: it may hold run identifiers, cursors, site references, and record counts, but never a clinical value, a subject identifier that is itself PHI, or any payload fragment. Those two boundaries are what make the orchestrator safe to place inside a GxP environment.

Building the DAG: Idempotent Tasks, Sensors, and Per-Site Mapping #

The core pattern is a DAG whose tasks are individually idempotent, whose fan-out is dynamic per site, and whose data never crosses XCom. The extract task returns only a reference — a URI and a content hash pointing at an external object store — and downstream tasks read the payload from that store, never from XCom. Dynamic task mapping (.expand()) produces one mapped instance per site so a single slow or failing site cannot block the others, and each mapped instance is scoped to the run’s deterministic data interval.

# 21 CFR Part 11 relevance: every task is attributable to one run_id and one
# data_interval; XCom carries only references (URI + SHA-256), never PHI.
from __future__ import annotations
import pendulum
from airflow.decorators import dag, task
from airflow.sensors.python import PythonSensor
from mypipeline import edc, store, audit, config

DEFAULTS = dict(
    owner="clinical-data-eng",
    retries=3,
    retry_delay=pendulum.duration(minutes=2),
    retry_exponential_backoff=True,
    max_retry_delay=pendulum.duration(minutes=20),
    depends_on_past=False,
)

@dag(
    dag_id="edc_sync_study_ONC2024",
    schedule="0 */4 * * *",              # every 4 hours
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,                        # never storm-backfill on deploy
    default_args=DEFAULTS,
    tags=["edc", "gxp", "study:ONC2024"],
)
def edc_sync():

    def _source_ready(**ctx) -> bool:
        # Readiness = the vendor's batch-close marker for THIS interval exists.
        return edc.batch_marker_present(
            study="ONC2024", interval_end=ctx["data_interval_end"])

    ready = PythonSensor(
        task_id="await_source_ready",
        python_callable=_source_ready,
        mode="reschedule",                # free the worker slot between pokes
        poke_interval=300, timeout=60 * 60,
    )

    @task
    def list_sites() -> list[str]:
        # Cursors/keys only — safe for XCom and the metadata DB.
        return config.active_sites(study="ONC2024")

    @task(retries=3)
    def extract(site: str, data_interval_start=None, data_interval_end=None) -> dict:
        # Extract is scoped to the deterministic window and writes to an
        # external store; only a reference (never the payload) is returned.
        ref = edc.extract_site(
            study="ONC2024", site=site,
            since=data_interval_start, until=data_interval_end)
        return {"site": site, "uri": ref.uri, "sha256": ref.sha256,
                "records": ref.count}

    refs = extract.expand(site=list_sites())   # one mapped instance per site
    audit.on_success >> None
    return ready >> refs

The extract task takes site from the mapped list and the interval bounds from the run context, so re-running any single mapped instance re-reads exactly the same window — the foundation of idempotency covered in depth in the companion deep-dive on Writing Idempotent Airflow DAGs for EDC Sync. The sensor runs in reschedule mode so a study that is not yet ready does not hold a worker slot hostage for an hour, and its timeout converts an indefinitely-late feed into a definite failure that alerting can act on. The retry_exponential_backoff with a max_retry_delay ceiling gives transient EDC API faults room to clear without hammering the vendor — the same discipline as Handling API Rate Limits in Clinical Sync and Async Polling Strategies for EDC Updates.

Compliance Constraints: Secrets, XCom Hygiene, and SLA Alerting #

The orchestrator introduces two failure surfaces that ordinary scripts do not: a shared metadata database that persists task return values, and a Connections store that can leak credentials. Both must be closed off explicitly. Credentials resolve through a secrets backend — never a hard-coded Connection or an environment variable baked into an image — and every task guards against accidentally serializing a payload into XCom. The transform, load, and audit stages below complete the DAG while enforcing those constraints, and a sla plus an on_failure_callback turn a late or failed safety feed into an actionable alert.

# 21 CFR Part 11 §11.10(d): system access is limited; secrets resolve from a
# backend at runtime. §11.10(e): a late/failed run raises an attributable alert.
from airflow.decorators import task
from airflow.exceptions import AirflowException
from mypipeline import store, transform as xf, warehouse, audit, alerting

@task(retries=3)
def validate(refs: list[dict]) -> list[dict]:
    for ref in refs:
        blob = store.read(ref["uri"])                 # payload from external store
        if store.sha256(blob) != ref["sha256"]:       # integrity gate
            raise AirflowException(f"hash mismatch for {ref['site']}")
        if not xf.schema_valid(blob):                 # ODM structural check
            raise AirflowException(f"invalid ODM for {ref['site']}")
    return refs                                       # references pass through

@task(retries=3)
def load(refs: list[dict], **ctx) -> dict:
    # Idempotent upsert on natural keys; safe to retry, safe to backfill.
    written = warehouse.upsert_from_refs(refs, run_id=ctx["run_id"])
    return {"run_id": ctx["run_id"], "rows": written}

def _alert_on_failure(context) -> None:
    # Fires on any task failure, sensor timeout, or SLA miss.
    alerting.page(
        study="ONC2024",
        task=context["task_instance"].task_id,
        run_id=context["run_id"],
        reason=str(context.get("exception", "sla_or_task_failure")))

# The load task carries an SLA so a late safety feed is flagged, not silent.
load_task = load.override(
    sla=pendulum.duration(minutes=30),
    on_failure_callback=_alert_on_failure,
)

Three properties make this inspection-ready. The hash gate in validate proves the bytes that reach transform are the bytes extraction recorded, closing the lineage loop. The secrets backend means a Connection URI is never committed and never printed, satisfying access-control expectations. And the SLA on load — not on the sensor — measures the metric a clinical reviewer actually cares about: whether ingested data landed in time for signal detection, with a breach routed straight to _alert_on_failure. When a batch cannot be recovered by retries, route it to the pattern in Dead-Letter Queues and Error Recovery rather than letting the DAG fail silently.

Configuration and Parameterization #

DAG behavior must be data, not code: schedules, SLAs, site lists, and connection identifiers differ per study and per environment, and any change to them is a change-controlled event. Externalize the policy into a validated file and map every secret through the secrets backend so the same DAG artifact is promoted unchanged from validation to production. The default_args block is the heart of the contract — it sets the retry, backoff, and ownership policy every task inherits.

# config/dags/edc_sync_ONC2024.yaml — version-controlled; changes require a tracked CR.
# GxP: this file is a configuration item; its git SHA is stamped into each run's audit header.
dag:
  dag_id: edc_sync_study_ONC2024
  schedule: "0 */4 * * *"
  start_date: "2026-01-01T00:00:00Z"
  catchup: false                       # controlled backfills only, via CLI
  max_active_runs: 1                    # no overlapping runs of the same study
default_args:
  owner: clinical-data-eng
  retries: 3
  retry_delay_minutes: 2
  retry_exponential_backoff: true
  max_retry_delay_minutes: 20
  depends_on_past: false
sla:
  load_minutes: 30
readiness:
  poke_interval_s: 300
  timeout_s: 3600
  mode: reschedule
connections:                            # names resolved via the secrets backend
  edc_api: edc_onc2024                  # NOT a URI — a key into Vault/SSM
  warehouse: wh_clinical_prod

The environment and connection mapping keeps credentials out of the artifact entirely. Each logical name maps to a secret path the backend resolves at runtime:

# pydantic enforces the DAG config schema so an invalid schedule or missing
# connection key fails fast in CI, not silently at scheduler parse time.
from pydantic import BaseModel, Field

class DefaultArgs(BaseModel):
    owner: str
    retries: int = Field(ge=0, le=10)
    retry_delay_minutes: float = Field(gt=0)
    retry_exponential_backoff: bool = True
    max_retry_delay_minutes: float = Field(gt=0)

class DagConfig(BaseModel):
    dag_id: str
    schedule: str
    catchup: bool = False
    max_active_runs: int = Field(default=1, ge=1)
    default_args: DefaultArgs

Two rules keep this auditable. The configuration file is a controlled item — its git SHA is stamped into each run’s audit header so a reviewer can prove which schedule and SLA policy governed a given run. And no credential ever lives in the file or the DAG code: connection keys name a secret path, and the secrets backend supplies the value at runtime, keeping the DAG promotable across the segregated environments the clinical data architecture and EDC standards program requires.

Testing and Validation #

A DAG is validated software, so it needs two distinct test tiers: DAG-integrity tests that prove the graph itself is well-formed (it imports, has no cycles, and every task carries a retry and owner policy), and task unit tests that prove each callable is correct and idempotent in isolation. The integrity tests run as a compliance gate in CI so a malformed DAG can never reach the scheduler, and their output is retained as an OQ artifact.

# OQ artifact: proves every shipped DAG imports cleanly, is acyclic, and carries
# the mandated retry/owner policy — the properties an inspector re-derives.
import pytest
from airflow.models import DagBag

@pytest.fixture(scope="module")
def dagbag():
    bag = DagBag(dag_folder="dags/", include_examples=False)
    return bag

def test_no_import_errors(dagbag):
    assert dagbag.import_errors == {}, dagbag.import_errors

def test_dags_are_acyclic_and_have_policy(dagbag):
    assert dagbag.dags, "no DAGs discovered"
    for dag in dagbag.dags.values():
        dag.test_cycle()                              # raises on any cycle
        for task in dag.tasks:
            assert task.retries >= 1, f"{task.task_id} has no retries"
            assert task.owner and task.owner != "airflow", task.task_id
        assert dag.catchup is False, f"{dag.dag_id} must not catchup on deploy"

# OQ artifact: proves the extract task is idempotent — re-running the same
# (site, interval) yields the same reference hash and never touches XCom payload.
from mypipeline.dags.edc_sync import extract
import pendulum

def test_extract_is_idempotent(monkeypatch, fake_edc):
    interval_start = pendulum.datetime(2026, 6, 27, tz="UTC")
    interval_end = interval_start.add(hours=4)
    kw = dict(data_interval_start=interval_start, data_interval_end=interval_end)

    first = extract.function(site="001", **kw)
    second = extract.function(site="001", **kw)

    assert first["sha256"] == second["sha256"]        # deterministic window
    assert "records" in first and "uri" in first
    assert "payload" not in first                      # never carries PHI in XCom

Mock fixtures should cover at minimum: a readiness sensor that times out, a mapped extract where one site fails while others succeed, a validate hash mismatch, and a load re-run that inserts zero new rows. Each test’s output is retained as objective evidence for the OQ protocol, and the test_no_import_errors and cycle checks gate every merge so a regression in DAG structure blocks deployment under change management.

Production Gotchas and Failure Modes #

• Catchup storm on first deploy. A DAG with catchup=True and a start_date months in the past schedules one run per missed interval the instant it is unpaused, overwhelming the EDC API and the scheduler. Remediation: ship every DAG with catchup=False and max_active_runs=1; perform any genuine historical load as a deliberate, rate-limited airflow dags backfill --reset-dagruns under a tracked change request.
• PHI leaked into XCom or the metadata DB. Returning a DataFrame or an ODM payload from a task serializes it into the metadata database, which is backed up, replicated, and outside the PHI store’s access controls — a reportable breach. Remediation: tasks return only URIs, hashes, and counts; enable AIRFLOW__CORE__XCOM_BACKEND pointed at an access-controlled store if larger references are unavoidable, and add a CI check that fails on any task returning a payload type.
• Timezone / execution_date confusion. Treating logical_date as wall-clock time, or filtering the EDC extract on “now” instead of data_interval_start/data_interval_end, makes a re-run pull a different window than the original — destroying idempotency and reproducibility. Remediation: always derive extract bounds from the interval, set start_date in UTC, and never call datetime.now() inside a task.
• Scheduler duplicate runs. Running two schedulers without proper HA locking, or a task with a non-idempotent side effect, can double-load a batch when a run is retried or a scheduler failover occurs. Remediation: make every write an idempotent upsert on natural keys (see the idempotency deep-dive), set max_active_runs=1, and rely on Airflow 2’s row-level scheduler locking rather than home-grown coordination.
• Sensor pinning a worker slot. A poke-mode sensor waiting an hour for source readiness occupies an executor slot the whole time, starving real work. Remediation: run readiness sensors in mode="reschedule" so the slot is freed between pokes, and set a finite timeout so a never-ready feed fails loudly instead of hanging.

Compliance Checklist #

• No task returns PHI or a clinical payload; XCom and the metadata DB carry only cursors, keys, hashes, and counts (ALCOA+ Attributable, access control).
• Every clinical value lives in an external, access-controlled store keyed by URI + SHA-256, never in Airflow’s metadata database.
• All Connections and secrets resolve through a configured secrets backend; no credential appears in DAG code, the repo, or an image.
• Every task carries an explicit retries, retry_delay, backoff ceiling, and a non-default owner (attributable ownership).
• Each DAG ships with catchup=False and max_active_runs=1; historical loads run only as tracked, rate-limited backfills.
• Extract and load are idempotent — re-running a task instance for the same (site, data_interval) produces identical downstream state (Consistent, Complete).
• Data-readiness sensors run in reschedule mode with a finite timeout, and timeouts raise an attributable alert.
• Safety-relevant feeds carry an sla; SLA misses fire an on_failure_callback that pages an accountable owner.
• The DAG config file is under version control and its git SHA is recorded in each run’s audit header (change-management trigger).
• DAG-integrity tests (import + no cycles + policy) and task unit tests pass as a CI gate before any deployment.
• Each run writes a terminal audit record: run_id, logical_date, data_interval, per-site record counts, and config git SHA (Contemporaneous, Original).
• DAG deployment follows CSV/change control — reviewed, approved, and traceable to a versioned artifact before it reaches production.

Related #

• Writing Idempotent Airflow DAGs for EDC Sync — the deep-dive on scoping runs to a deterministic interval and upserting on natural keys.
• Async Polling Strategies for EDC Updates — the polling cadence a readiness sensor coordinates with.
• Dead-Letter Queues and Error Recovery — where a batch goes when retries are exhausted.
• Handling API Rate Limits in Clinical Sync — the backoff discipline each extract task honors against the vendor quota.
• Automated EDC Ingestion & Sync Pipelines — the parent reference for this discipline.