Orchestrating Clinical ETL with Prefect

Prefect 2.x turns a clinical FHIR/HL7 batch pipeline into ordinary Python: the DAG is not a static graph you declare up front but the actual call tree the interpreter walks at runtime, so the number of Patient batches, the retry policy per resource type, and the concurrency cap that protects your terminology server are all decided from data rather than from a hand-drawn topology. This page sits within Async Batch Processing for Large Datasets and gives the Prefect-specific implementation: fanning out over FHIR resource batches with .map(), bounding parallelism so a retry storm never overruns a fragile $validate-code endpoint, and persisting task results so a retried extract never double-loads a Patient.

The precise problem this page solves: how do you orchestrate ingestion of tens of thousands of FHIR resources on Prefect so that transient 429/5xx failures self-heal through idempotent retries, fan-out parallelism is capped by a work-pool limit rather than by batch count, and a full re-run converges to exactly the same target state as the first run — without ever writing PHI into task results or flow state?

Prefect construct → clinical-ETL role (quick reference)

Prefect’s orchestration primitives map cleanly onto the moving parts of a clinical batch load. Keep this table next to your flow definition — most Prefect ETL bugs are a misused primitive (a retry that is not idempotent, an unbounded .map()) rather than broken transform logic.

The architectural rule behind the table: Prefect is an orchestration layer, not a data-processing engine. Task return values (which land in the result store and flow state) carry references — batch IDs, counts, content hashes — never Patient.name, identifiers, or raw bundles. The same NDJSON batch shape produced by a bulk data export is the canonical input assumed below.

Contrast with the Airflow approach

The sibling Airflow implementation solves the identical problem with a different model, and the difference is worth internalizing before you commit to a framework. Airflow declares a static DAG of Operator instances resolved at parse time; dynamic fan-out arrives late via expand(), cross-task data must be tiny because it flows through XCom in the metadata database, and asynchronous work needs a deferrable operator and a triggerer. Prefect inverts this: the flow is imperative Python, so batches are discovered by simply calling code, .map() fan-out is first-class and dynamic, async def tasks run natively on the event loop, and results move through a pluggable result store rather than a shared database. For a clinical pipeline whose batch count depends on how many patients a nightly export produced, Prefect’s runtime graph is the more natural fit; for a fixed, schedule-driven topology with heavy operator reuse, Airflow’s static DAG is easier to reason about. Both must still be idempotent — the idempotent clinical data loads discipline is framework-independent.

Implementation pattern

The flow below is the complete worker: it lists Patient-ID batches, maps an idempotent extract-and-transform task over them under a concurrency cap, and commits with a single load task. Note that every task return value is a reference (batch ID, row count, content hash) — never a raw resource — so nothing PHI-bearing ever reaches the result store.

import hashlib
from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash

# A named concurrency limit created once with the Prefect API/CLI:
#   prefect concurrency-limit create fhir-tx 8
# caps how many tasks holding this tag run at once, protecting the
# FHIR/terminology server no matter how wide .map() fans out.
FHIR_TX_TAG = "fhir-tx"


def _stable_batch_key(context, arguments) -> str:
    """Cache key from the batch id + resource type only (never PHI)."""
    batch = arguments["batch"]
    return hashlib.sha256(
        f"{batch['resource_type']}:{batch['batch_id']}".encode()
    ).hexdigest()


@task(
    retries=3,
    retry_delay_seconds=[10, 30, 90],   # explicit exponential backoff
    tags=[FHIR_TX_TAG],                  # subject to the concurrency limit
    persist_result=True,                 # write result to the result store
    cache_key_fn=_stable_batch_key,      # re-run reuses the prior result
)
def extract_transform(batch: dict, client) -> dict:
    """Extract one batch of Patients, transform to load-ready rows.

    Idempotent: keyed on batch_id, produces the same output for the same
    input, and returns only references + a content hash — no PHI.
    """
    log = get_run_logger()
    # A transient 429/5xx here raises; Prefect retries with backoff.
    resources = client.read_patients(batch["patient_ids"])
    rows = [to_load_row(r) for r in resources]
    payload_hash = hashlib.sha256(
        repr(sorted(r["patient_id"] for r in rows)).encode()
    ).hexdigest()
    log.info("batch %s -> %d rows", batch["batch_id"], len(rows))  # ids/counts only
    return {"batch_id": batch["batch_id"], "rows": rows, "hash": payload_hash}

The @task decorator does the heavy lifting. retries=3 with retry_delay_seconds=[10, 30, 90] gives explicit exponential backoff — a list schedules each successive retry further out, so a rate-limited FHIR server recovers instead of being hammered. The fhir-tx tag ties the task to a named concurrency limit, and cache_key_fn computes a stable key from batch_id and resource_type so a retry or a whole re-run reuses the persisted result rather than re-reading the server. Because the key is derived from identifiers you control, it is deterministic across runs — the foundation of idempotency.

@task
def load(results: list[dict], writer) -> dict:
    """Terminal load: idempotent upsert keyed on a stable patient id."""
    log = get_run_logger()
    total = 0
    for res in results:
        for row in res["rows"]:
            # UPSERT (ON CONFLICT DO UPDATE) on the stable natural key,
            # so a retried or re-run batch never inserts a duplicate.
            writer.upsert(key=row["patient_id"], values=row)
            total += 1
    log.info("loaded %d rows across %d batches", total, len(results))
    return {"loaded_rows": total, "batches": len(results)}


@flow(name="clinical-fhir-batch-load")
def clinical_etl(as_of: str):
    """List patient batches and fan out under a concurrency cap."""
    client, writer = build_fhir_client(), build_writer()
    batches = list_patient_batches(as_of)          # N decided from data at runtime
    # .map() fans out one task per batch; the fhir-tx limit caps how many
    # run concurrently regardless of how many batches exist.
    futures = extract_transform.map(batches, client=unmapped(client))
    results = [f.result() for f in futures]
    return load(results, writer)

The load task performs an UPSERT on the stable natural key (here patient_id), so re-running the flow — or Prefect retrying a mapped task after a partial write — converges to the same target state instead of inserting duplicates. .map() returns a list of futures; the flow resolves them and hands the surviving results to a single terminal load. unmapped(client) passes the shared client to every mapped call without treating it as a per-batch argument.

Validation and testing

Idempotency and retry behavior are the two properties worth proving in CI, and both are testable against a stub without a live FHIR server. Run the flow twice against an in-memory writer and assert the second run adds no rows; inject a transient error and assert the retry count fires.

def test_reruns_are_idempotent():
    """A second run over the same batches loads zero new rows."""
    writer = InMemoryWriter()           # upsert keyed on patient_id
    client = StubFhirClient(PATIENTS)   # deterministic source

    clinical_etl(as_of="2026-07-15", writer=writer, client=client)
    first = writer.row_count
    clinical_etl(as_of="2026-07-15", writer=writer, client=client)

    # Re-run upserts the same keys; no duplicates appear.
    assert writer.row_count == first


def test_transient_error_triggers_retry():
    """A 429 on the first attempt succeeds on retry, not a hard failure."""
    client = FlakyFhirClient(fail_times=2, error=TransientHttpError(429))
    result = extract_transform.fn(  # .fn calls the wrapped function directly
        batch={"resource_type": "Patient", "batch_id": "b1",
               "patient_ids": ["p1"]},
        client=client,
    )
    assert result["batch_id"] == "b1"

Test the pure logic with .fn (the undecorated function) for speed, and reserve full-flow runs for the idempotency assertion where orchestration actually matters. Because the cache key and upsert key are both derived from stable identifiers, the idempotency test also guards against an accidental change to either — flip the key to a timestamp and the second-run assertion fails immediately.

Gotchas and compliance constraints

• Never let PHI reach a task result or state. Prefect persists return values to the result store and records state in its backend; a task that returns a raw Patient bundle writes Protected Health Information into orchestration infrastructure and task-run logs. Return references, counts, and hashes only, and scrub identifiers before any logger call — the same log-hygiene rule enforced in the Airflow implementation.
• A retry is only safe if the task is idempotent. retries= will re-execute a failed task; if extract_transform had already committed a partial write with INSERT semantics, the retry doubles the load. Anchor every write on an UPSERT keyed on a stable natural id and derive cache_key_fn from that same id, so retries and re-runs both converge. Distributed variants face the identical hazard — see distributed FHIR processing with Apache Spark.
• Cap concurrency so backpressure protects the server, not the other way around. An unbounded .map() over 20,000 batches launches 20,000 tasks that stampede a $validate-code or FHIR REST endpoint and trigger cascading 429s. Bind the fan-out to a named concurrency limit or work-pool limit sized to what the terminology server tolerates, so the number of resource batches never determines the load on the server.

Deployment checklist

• Named concurrency limit created (prefect concurrency-limit create fhir-tx N) and applied via a task tag.
• Retries with explicit backoff (retries=, retry_delay_seconds=[...]) on every task that calls the FHIR/terminology server.
• Idempotent writes — every load is an UPSERT on a stable natural key; cache_key_fn derives from the same key.
• PHI kept out of results and state — tasks return references/counts/hashes; identifiers scrubbed before logging.
• Idempotency regression test — a double run asserts zero duplicate rows in CI.

By modeling the pipeline as imperative Python, Prefect lets a clinical ETL team fan out dynamically over however many resource batches an export produced while keeping retries idempotent, concurrency bounded, and PHI out of every orchestration surface — the same guarantees the Airflow and Spark siblings reach by other means.

Related

• Scaling FHIR batch processing with Apache Airflow — the static-DAG counterpart to this flow, with the same PHI-isolation rules
• Implementing idempotent clinical data loads — the upsert and stable-key discipline that makes retries safe
• Distributed FHIR processing with Apache Spark — scaling the same fan-out beyond a single Prefect worker
• Async Batch Processing for Large Datasets — parent overview of the batch orchestration options