Lesson 1 - Designing the DAG

Welcome to Designing the DAG#

Every DAG in Modules 2-4 was built to teach one idea — a sensor, a hook, a retry. This module builds something closer to what CityFlow’s real pipeline needs to be: a DAG shaped like production ETL. Before writing any of the real logic, this lesson asks the question that decides everything else: where do the task boundaries go, and why there specifically?

By the end of this lesson, you will be able to:

• State the real reason ingest, validate, transform, and load are four separate tasks instead of one script or two
• Predict what happens to downstream tasks when an upstream task fails, and confirm it with a real DAG run
• Read upstream_failed in a real Grid view and explain what it actually means

Four Tasks, Not One Script#

It would be shorter to write CityFlow’s ETL pipeline as a single Python function: fetch the data, check it, reshape it, write it to Postgres, done. Airflow would happily run that as one task. The reason this module doesn’t do that comes down to one question for each boundary: what should happen if this step’s output turns out to be bad, and does the next step deserve to find out before it runs, or after it’s already burned time and resources finding out itself?

• Ingest is separate from validate because ingest’s job is just “get the data here” — it has no opinion about whether the data is any good. Mixing a data-quality decision into the same task that’s still busy reading a file makes both harder to reason about and re-run independently.
• Validate is separate from transform for the sharpest reason in this list: a validation failure should stop the DAG before transform ever starts. If validate and transform were the same task, a bad batch would mean transform’s aggregation logic runs anyway — on data validate already knows is broken — burning real compute on work whose result nobody should trust. Two tasks means Airflow’s own dependency graph enforces that: transform simply doesn’t start unless validate succeeded.
• Transform is separate from load because transform is pure computation — no network calls, nothing that can fail because a database happens to be unreachable — while load is the one task in the whole pipeline that talks to an outside system. Isolating that means a Postgres outage only ever fails the load task, not the (already-finished, already-correct) transform work sitting right behind it.

This is the same idea Module 3’s retry lesson touched on: a task boundary is a unit of both retry and blast radius. Four tasks means four independent places to retry, four independent log streams, and — the part this lesson actually proves — four independent points where a real failure stops real downstream work from running at all.

The Skeleton, Wired the Right Way#

# gate: skip
"""CityFlow's real ETL skeleton (Module 5 Lesson 1).

Four stub tasks, wired ingest -> validate -> transform -> load, standing in
for the real logic Lessons 2-4 build. The point of this DAG isn't what each
stub does (nothing) -- it's the *shape*: validate sits between ingest and
transform on purpose, so a validation failure stops the DAG before any
transform work runs. dag_run.conf's "force_validation_failure" flag lets the
same file demonstrate both the happy path and that stop-before-waste
behavior for real, against the live scheduler."""
from datetime import datetime

from airflow.decorators import dag, task
from airflow.exceptions import AirflowException


@dag(
    dag_id="cityflow_etl_skeleton",
    schedule=None,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["cityflow", "module-5", "etl-skeleton"],
)
def cityflow_etl_skeleton_dag():
    @task
    def ingest_stub() -> str:
        print("ingest_stub: would fetch CityFlow's raw taxi data here")
        return "raw_trips_placeholder"

    @task
    def validate_stub(raw_ref: str, **context) -> str:
        force_fail = context["dag_run"].conf.get("force_validation_failure", False)
        print(f"validate_stub: checking {raw_ref!r}, force_validation_failure={force_fail}")
        if force_fail:
            raise AirflowException(
                "validate_stub: simulated data-quality failure -- stopping "
                "the DAG here on purpose, before transform_stub or load_stub "
                "ever run and waste work on data that didn't pass."
            )
        return "clean_trips_placeholder"

    @task
    def transform_stub(clean_ref: str) -> str:
        print(f"transform_stub: would aggregate {clean_ref!r} here")
        return "transformed_placeholder"

    @task
    def load_stub(transformed_ref: str) -> None:
        print(f"load_stub: would write {transformed_ref!r} to Postgres here")

    load_stub(transform_stub(validate_stub(ingest_stub())))


cityflow_etl_skeleton_dag()

Nothing here does real work yet — that’s on purpose. validate_stub accepts the DAG run’s own conf dict via **context, and reads a flag, force_validation_failure, that lets this exact file demonstrate both outcomes for real: the normal path, and the one this lesson is actually about.

Triggered Twice: the Happy Path, Then the Point#

# gate: skip
import json
import subprocess
import time

dag_id = "cityflow_etl_skeleton"


def trigger_and_wait(run_id, conf=None, expect="success"):
    cmd = ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "trigger", dag_id, "--run-id", run_id]
    if conf:
        cmd += ["--conf", json.dumps(conf)]
    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode != 0:
        subprocess.run(["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "reserialize"],
                        capture_output=True, text=True, check=True)
        subprocess.run(cmd, capture_output=True, text=True, check=True)
    print(f"triggered {run_id!r}")

    deadline = time.time() + 90
    state = "queued"
    while state not in ("success", "failed") and time.time() < deadline:
        time.sleep(3)
        out = subprocess.run(
            ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "list-runs", "-d", dag_id, "--output", "json"],
            capture_output=True, text=True, check=True,
        )
        runs = json.loads(out.stdout)
        match = [x for x in runs if x["run_id"] == run_id]
        state = match[0]["state"] if match else "queued"
    print(f"run {run_id!r} reached state: {state}")
    assert state == expect, f"expected {expect}, got {state}"

The first trigger is the happy path — no conf, every stub runs, every stub succeeds:

triggered 'lesson1_m5_happy_1784913140'
run 'lesson1_m5_happy_1784913140' reached state: success
dag_id                | execution_date            | task_id        | state   | start_date                       | end_date
======================+===========================+================+=========+==================================+=================================
cityflow_etl_skeleton | 2026-07-24T17:12:26+00:00 | ingest_stub    | success | 2026-07-24T17:19:55.906375+00:00 | 2026-07-24T17:19:55.983177+00:00
cityflow_etl_skeleton | 2026-07-24T17:12:26+00:00 | validate_stub  | success | 2026-07-24T17:19:56.142449+00:00 | 2026-07-24T17:19:56.218102+00:00
cityflow_etl_skeleton | 2026-07-24T17:12:26+00:00 | transform_stub | success | 2026-07-24T17:19:57.216688+00:00 | 2026-07-24T17:19:57.294140+00:00
cityflow_etl_skeleton | 2026-07-24T17:12:26+00:00 | load_stub      | success | 2026-07-24T17:19:57.844255+00:00 | 2026-07-24T17:19:57.909833+00:00

The second trigger passes {"force_validation_failure": true} as the DAG run’s conf — this is the run that actually proves the design argument:

triggered 'lesson1_m5_forcefail_1'
run 'lesson1_m5_forcefail_1' reached state: failed
dag_id                | execution_date            | task_id        | state           | start_date                       | end_date
======================+===========================+================+=================+==================================+=================================
cityflow_etl_skeleton | 2026-07-24T17:21:18+00:00 | ingest_stub    | success         | 2026-07-24T17:21:19.542583+00:00 | 2026-07-24T17:21:19.618112+00:00
cityflow_etl_skeleton | 2026-07-24T17:21:18+00:00 | validate_stub  | failed          | 2026-07-24T17:21:20.581313+00:00 | 2026-07-24T17:21:20.652174+00:00
cityflow_etl_skeleton | 2026-07-24T17:21:18+00:00 | transform_stub | upstream_failed | 2026-07-24T17:21:20.697856+00:00 | 2026-07-24T17:21:20.697856+00:00
cityflow_etl_skeleton | 2026-07-24T17:21:18+00:00 | load_stub      | upstream_failed | 2026-07-24T17:21:21.555155+00:00 | 2026-07-24T17:21:21.555155+00:00

ingest_stub still succeeded — ingest doesn’t care about data quality, it already did its job. validate_stub genuinely failed, raising the real AirflowException it was written to raise. And transform_stub and load_stub never ran at all: their start_date and end_date are identical, because Airflow’s default trigger rule (all_success) marked them upstream_failed the instant validate_stub failed, without ever handing either one to a worker. This is the whole design argument, made real: separating validate from transform doesn’t just look cleaner on a diagram, it’s the mechanism that stopped two tasks from running on data already known to be bad.

The task’s own log confirms the same thing from inside the failed task itself:

[2026-07-24T17:21:20.647+0000] {logging_mixin.py:190} INFO - validate_stub: checking 'raw_trips_placeholder', force_validation_failure=True
[2026-07-24T17:21:20.650+0000] {taskinstance.py:3311} ERROR - Task failed with exception
airflow.exceptions.AirflowException: validate_stub: simulated data-quality failure -- stopping the DAG here on purpose, before transform_stub or load_stub ever run and waste work on data that didn't pass.

A Real Screenshot of the Stopped Run#

Orange is a different color from red on purpose. Red means a task ran and genuinely failed. Orange means Airflow looked at the dependency graph, saw an upstream failure, and decided not to run the task in the first place — exactly the “stop before wasting work” behavior this lesson set out to prove.

Key Takeaways#

• Every task boundary in an ETL DAG is a real decision about retry granularity and blast radius, not just a stylistic split — validate sits between ingest and transform specifically so a bad batch stops before transform runs on it.
• A DAG run’s conf dict (context["dag_run"].conf), read inside a task via **context, is a real way to change a task’s behavior per-trigger without editing the DAG file — this lesson used it to make the same file demonstrate two different real outcomes.
• upstream_failed (orange) is not the same as failed (red): a failed task actually ran and raised; an upstream_failed task was never given to a worker because Airflow’s default all_success trigger rule already knew it couldn’t succeed.
• This module’s remaining lessons fill in these four stubs with real logic one at a time — the shape proven here doesn’t change, only what’s inside each box.

Exercises#

Exercise 1 — Change validate_stub’s trigger rule discussion into code: add a fifth stub task, alert_stub, with trigger_rule="one_failed", wired to run only when at least one upstream task fails. Trigger the forced-failure run again and confirm alert_stub reaches success while transform_stub and load_stub still show upstream_failed.

Exercise 2 — Read context["dag_run"].conf in ingest_stub too, and add a second flag, simulate_ingest_failure, that makes ingest_stub itself raise. Trigger with only that flag set and confirm all three downstream stubs (validate_stub, transform_stub, load_stub) show upstream_failed — the same mechanism, one boundary earlier.

Exercise 3 — Use the Airflow UI’s Graph view (not Grid) for the forced-failure run and compare it side by side with the Grid view above — which one makes the shape of the stoppage easier to see at a glance, and which makes the timing (start/end dates) easier to read? Write one sentence on when you’d reach for each view in a real incident.