Module 13: Batch ETL Pipeline Orchestration

Why this matters

Module 03 taught you to schedule a single recurring job with Celery Beat — "run this task every night at 2am." Real data pipelines are rarely one task: a nightly ETL run typically extracts from three source tables, transforms each independently, joins them, loads the result into a warehouse (the OLAP module in track 04), and every one of those steps depends on the ones before it succeeding. If the extract from one source fails, running the transform on stale or missing data doesn't just waste compute — it can silently corrupt a report someone makes a real decision from. This module covers dependency-aware pipeline orchestration: a purpose-built tool (this module uses Prefect) that understands which steps depend on which, blocks downstream steps when an upstream step fails, retries individual steps without restarting the whole pipeline, and lets you safely re-run a single historical day's data (a backfill) without touching any other day — none of which a bare cron schedule or a hand-composed sequence of Celery tasks gives you for free.

Concepts

Why this isn't just "another Celery Beat schedule"

Module 03's Celery Beat handles when a job runs. Module 01's Celery chains/groups/chords handle composing multiple tasks into a sequence or fan-out. A dedicated pipeline orchestrator like Prefect combines both concerns and adds the two things neither of those gives you cleanly: a persisted, inspectable record of every step's success/failure across runs (so you can actually see "did last Tuesday's load succeed?"), and first-class support for re-running one specific historical run (a backfill) independently of the live schedule. You could hand-build all of this on top of Celery primitives — teams did, for years — but a pipeline orchestrator exists because that hand-built version is a recurring, solved problem, not a novel one.

Tasks and flows: naming the same shape you already know

The vocabulary maps directly onto what module 01 already taught:

A task is one unit of work — extract, transform, or load one piece — the same idea as a Celery task.
A flow is the orchestration function that calls tasks and defines their dependencies through plain function calls and argument passing — if task B's function call uses task A's return value, B depends on A, and the orchestrator understands that without you declaring it separately.

from prefect import flow, task

@task
def extract():
    return [1, 2, 3, 4, 5]

@task
def transform(data):
    return [x * 2 for x in data]

@task
def load(data):
    return sum(data)

@flow
def etl_pipeline():
    raw = extract()
    transformed = transform(raw)   # depends on `raw` -> depends on extract()
    return load(transformed)       # depends on `transformed` -> depends on transform()

Failure blocks downstream automatically — no manual checking required

If extract() raises, transform() and load() never run — the orchestrator sees that transform's call depends on extract's result and doesn't attempt it once that dependency has failed. You don't write if extract_succeeded: transform(...) anywhere; the dependency, expressed as an ordinary function call, is the whole mechanism.

  extract() FAILS
       │
       ▼
  transform() never runs   (its input never existed)
       │
       ▼
  load() never runs        (same reason, one level further)

  Compare to a naive script that "runs the next step regardless" --
  that script would call transform(None) or transform(garbage) and
  either crash confusingly or, worse, silently produce wrong output.

Per-task retries — module 02's discipline, declared once per task

Just like module 02's Celery retry decorator and the durable-workflow engine's retry policy (module 08, distributed-systems track), a task can declare its own retry policy, and only that task retries — a transient failure in one extract doesn't restart the whole pipeline from scratch:

@task(retries=3, retry_delay_seconds=30)
def extract_from_flaky_api():
    ...  # only this task's failures are retried, up to 3 times

Backfills: safely re-running one historical partition

A pipeline that processes "yesterday's data" every night needs a way to re-run a specific past day — say, after fixing a bug that corrupted last Tuesday's numbers — without touching any other day's data or re-running the live nightly schedule early. The standard technique is parameterizing the flow by the date/partition it processes, so calling it for a historical date is just calling the same flow with a different argument:

@flow
def daily_etl(run_date: str):
    data = extract_for_date(run_date)
    load_for_date(data)

# The live schedule calls this with today's date.
# A backfill calls it explicitly for a past date -- same code, different input:
daily_etl(run_date="2026-07-27")
daily_etl(run_date="2026-07-28")
daily_etl(run_date="2026-07-29")

For this to be safe, load_for_date must be idempotent per partition (module 01's idempotency lesson, applied to data pipelines): re-running daily_etl(run_date="2026-07-27") a second time should replace that day's data cleanly (e.g. DELETE ... WHERE date = :run_date then insert, or an upsert keyed on the date), not append a second copy alongside the first.

DAGs, not just chains: fan-out and fan-in

A real pipeline is rarely one straight line. Multiple independent extracts can run and feed a single downstream step that combines them — this is a DAG (directed acyclic graph), the same shape module 01's Celery chords gave you, expressed here as ordinary function calls:

@flow
def multi_source_etl():
    orders = extract_orders()      # independent
    refunds = extract_refunds()    # independent, runs regardless of orders' timing
    return combine(orders, refunds)  # depends on BOTH

combine only runs once both extract_orders and extract_refunds have completed — expressed with no explicit "wait for both" code, purely by combine's function signature needing both values.

Command reference

This module uses Prefect, a Python-native pipeline orchestrator that runs entirely locally for development (a lightweight embedded server starts automatically) — no separate infrastructure required for these exercises.

Concern	Prefect (Python)
Define a task	`@task` decorator on a function
Define a flow (the pipeline)	`@flow` decorator on a function
Express a dependency	Pass one task's return value as another task's argument
Retry a task automatically	`@task(retries=N, retry_delay_seconds=S)`
Run a flow	Call it like a normal Python function: `my_flow()`
Run for a specific historical partition (backfill)	Call the flow with an explicit parameter: `my_flow(run_date="2026-07-27")`

Hands-on exercises

pip install prefect. Everything below runs locally with no external services — Prefect starts a temporary local server automatically the first time a flow runs.

1. Run a linear ETL pipeline and see the dependency chain execute

from prefect import flow, task

@task
def extract():
    print("extracting raw data...")
    return [1, 2, 3, 4, 5]

@task
def transform(data):
    print("transforming...")
    return [x * 2 for x in data]

@task
def load(data):
    print("loading...")
    return sum(data)

@flow
def etl_pipeline():
    raw = extract()
    transformed = transform(raw)
    result = load(transformed)
    print("RESULT:", result)
    return result

if __name__ == "__main__":
    etl_pipeline()

Expected: extracting raw data..., transforming..., loading..., then RESULT: 30 — sum([2,4,6,8,10]). Each task's completion is logged with its own status.

2. Prove a failed step blocks everything downstream

from prefect import flow, task

@task
def extract():
    raise RuntimeError("source system unavailable")

@task
def transform(data):
    print("transform ran (should NOT happen)")
    return data

@flow
def etl_pipeline():
    raw = extract()
    transformed = transform(raw)
    return transformed

if __name__ == "__main__":
    try:
        etl_pipeline()
    except Exception as e:
        print("FLOW FAILED AS EXPECTED:", e)

Expected: extract fails and is logged as Failed; the string "transform ran" never prints — transform was never attempted because its input (extract's return value) never existed. This is the core dependency-aware behavior, proven by its absence, not asserted.

3. Prove per-task retries work without restarting the whole pipeline

from prefect import flow, task

attempt_count = {"n": 0}

@task(retries=3, retry_delay_seconds=1)
def flaky_extract():
    attempt_count["n"] += 1
    print(f"attempt {attempt_count['n']}")
    if attempt_count["n"] < 3:
        raise RuntimeError("transient failure")
    return "data-loaded"

@flow
def etl_pipeline():
    return flaky_extract()

if __name__ == "__main__":
    result = etl_pipeline()
    print("RESULT:", result, "total attempts:", attempt_count["n"])

Expected: attempt 1, attempt 2, attempt 3, then RESULT: data-loaded total attempts: 3 — the task retried itself automatically per its retries=3 policy, with no hand-written retry loop, exactly like module 02's Celery retries and module 08's (distributed-systems track) durable-workflow retry policy.

4. Run a fan-out/fan-in DAG

from prefect import flow, task

@task
def extract_orders():
    return [10, 20, 30]

@task
def extract_refunds():
    return [5]

@task
def combine(orders, refunds):
    return sum(orders) - sum(refunds)

@flow
def multi_source_etl():
    orders = extract_orders()
    refunds = extract_refunds()
    return combine(orders, refunds)

if __name__ == "__main__":
    print("RESULT:", multi_source_etl())

Expected: RESULT: 55 (10+20+30-5) — combine correctly waited for both independent extracts, expressed with nothing more than its function signature needing both values.

5. Run a parameterized backfill for three historical dates

from prefect import flow, task

processed_partitions = []

@task
def extract_for_date(run_date: str):
    print(f"extracting data for {run_date}")
    return {"run_date": run_date, "rows": 100}

@task
def load_for_date(data):
    processed_partitions.append(data["run_date"])
    print(f"loaded partition {data['run_date']}")

@flow
def daily_etl(run_date: str):
    data = extract_for_date(run_date)
    load_for_date(data)

if __name__ == "__main__":
    for d in ["2026-07-27", "2026-07-28", "2026-07-29"]:
        daily_etl(run_date=d)
    print("processed partitions:", processed_partitions)

Expected: each date is extracted and loaded independently, in order, and processed_partitions ends with all three dates — proving the same flow code handles both the live nightly run and an explicit historical backfill, just by varying the run_date argument.

6. Make the backfill idempotent and prove a re-run doesn't duplicate data

Change load_for_date to write into a dict keyed by run_date instead of appending to a list:

warehouse = {}

@task
def load_for_date(data):
    warehouse[data["run_date"]] = data["rows"]  # overwrite, not append

Run daily_etl(run_date="2026-07-27") twice in a row. Expected: warehouse["2026-07-27"] holds one value, not a growing list or a doubled count — re-running the same partition replaced its data cleanly instead of duplicating it, exactly the property a real backfill needs (module 01's idempotency lesson, applied to a data partition instead of a single task invocation).

7. Diagnose and fix: a corrupted report nobody noticed for a week

A nightly pipeline extracts from three source tables, transforms each, and loads a combined report. One source's extract has been silently failing for a week — but the team's original hand-rolled script (built before adopting a real orchestrator) just logged the error and let the transform/load steps run anyway with whatever partial data it had. The weekly business report has been wrong for a week with no alert.

Solution

Root cause: the hand-rolled pipeline didn't treat a failed extract as a reason to block the rest of the pipeline — it logged and continued, producing a report from incomplete data that looked plausible enough that nobody caught it. This is exactly the gap module 01's Celery chains and a bare cron schedule don't close on their own: nothing enforces that a downstream step can't run without its input actually succeeding unless you write that check yourself, everywhere, correctly, forever.

Fix: express the pipeline as a proper dependency graph (as this module's flows do) so a failed extract's result never reaches the transform/load steps at all — they simply never run, and the flow's overall status shows Failed, which is what should have triggered an alert the moment it happened rather than a week of silently wrong reports. This is the concrete value of dependency-aware orchestration over "a script that calls things in order and hopes": the failure to run the rest of the pipeline is the safety mechanism, not a bug to work around.

8. Clean up

# No persistent services were started; Prefect's temporary local server
# stops automatically when the script exits.

Independent challenge

No code given. Design a nightly ETL pipeline that feeds the star schema from track 04's OLAP module: extract from orders, customers, and products source tables (three independent extracts), transform each into its corresponding dimension/fact shape, then load into the warehouse in the correct dependency order (dimensions before the fact table that references them). Specify: (1) the full dependency graph — which steps can run in parallel and which must wait for others; (2) which steps need a retry policy and roughly why (which ones call external/flaky systems vs. which are pure local transforms); (3) how you would safely backfill just the orders data for one specific past week without re-processing customers or products for that same window; (4) what should happen to the fact-table load step if the products extract fails that night.

Stuck? One hint

The three extracts (orders, customers, products) are independent and can run in parallel (fan-out); each has its own transform step depending only on its own extract; the fact-table load depends on all three transformed outputs being ready (fan-in), since it needs valid foreign keys into every dimension. Extracts (calling real source databases/APIs) deserve retries; pure in-memory transforms usually don't need them. A backfill parameterizes the flow by date range and re-runs only the orders extract/transform/load for that window, leaving customers/products flows completely untouched — they're separate flow parameters, not entangled in one all-or-nothing script. If products fails, the fact-table load must not run at all (per this module's core lesson) — better a late, correct report than an on-time, wrong one.

Common mistakes & troubleshooting

Letting a pipeline continue after a failed step "because the rest might still be useful." As the diagnose-and-fix exercise showed, this is exactly how a silently wrong report ships for a week. A failed dependency should block what depends on it, full stop.
Treating this as just a fancier cron job. Module 03's Beat schedule handles timing; this module's orchestrator handles dependencies between steps and safe historical re-runs — different problems that happen to often show up together in the same pipeline.
Non-idempotent backfills. Re-running a historical partition must replace that partition's data, not append a duplicate copy — exercise 6 demonstrated the fix (overwrite/upsert keyed by partition, not blind insert).
Retrying every step uniformly regardless of what it does. A pure in-memory transform that raises is a bug, not a transient condition — retrying it just delays surfacing the bug. Reserve retry policies for steps that call something genuinely flaky (an external API, a network call).
Forgetting the fan-in case. A step that depends on multiple upstream branches (like combine in exercise 4) must wait for all of them, not just the first to finish — verify this explicitly rather than assuming your orchestrator "just handles it" without checking.

Checkpoint quiz

Write down your answer to each question before expanding it — checking without attempting first is the single easiest way to fool yourself into thinking you've learned this.

What's the difference between what module 03's Celery Beat solves and what this module's pipeline orchestrator solves?
How does a task's dependency on another task get expressed in this module's flows, and what happens to a downstream task if its upstream dependency fails?
Why should retries typically be applied per-task rather than restarting the whole pipeline from the beginning on any failure?
What makes a backfill safe to run repeatedly for the same historical partition, and what breaks if that property is missing?
In a fan-in DAG where one step depends on two independent upstream branches, when does that step actually run?

Answers

Celery Beat (module 03) solves when a job runs — a schedule. This module's pipeline orchestrator solves dependencies between multiple steps (blocking downstream work on upstream failure) and safe historical re-runs (backfills) — concerns a bare schedule doesn't address on its own, even though a real pipeline usually needs both together.
A dependency is expressed simply by one task's function call using another task's return value as an argument — no separate dependency declaration is needed. If the upstream task fails, the downstream task that depends on its result never runs at all, because the value it needs was never produced.
Because a transient failure is usually local to one step (one flaky API call, say) — restarting the entire pipeline from scratch wastes the work every other, already-successful step already did, and delays detecting the real problem. A per-task retry policy retries only the step that actually failed.
The step that writes data for a given partition must overwrite or upsert that partition's data rather than blindly inserting/appending — the same idempotency requirement module 01 introduced for a single task, applied here to a whole historical partition. Without it, a re-run creates duplicate data instead of cleanly replacing what was there.
Only once both upstream branches have completed successfully — expressed by the step's function needing both values as arguments, with no manual "wait for both" logic required.