Lesson 3 - Your First DAG

Welcome to Your First DAG#

Every lesson so far has been about a stack that already existed. This one changes that: you’ll write a real DAG file, save it where Airflow actually looks, and watch a system you didn’t have to babysit find it, run it, and record that it succeeded — the exact promise Module 1 made about what an orchestrator adds over a hand-built script chain.

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

• Write a small DAG with the TaskFlow API (@dag and @task decorators) instead of hand-wiring operator objects
• Explain how a plain Python return value becomes data the next task receives, via XCom
• Confirm the scheduler has picked up a new DAG file, using airflow dags list
• Trigger a real DAG run and poll it to completion using the real Airflow CLI

The DAG: Three Tasks, One File#

cityflow_hello_dag.py does one small, real thing: it reads CityFlow’s actual NYC taxi-zone lookup table (the same 265-zone reference file static/datasets/nyc-taxi/taxi_zone_lookup.csv uses, copied into the mounted dags/data/ folder), checks that the count makes sense, and prints a greeting using it. No trip data yet — that’s the guided project. This lesson is about proving the mechanics work before asking Airflow to do more:

# gate: skip
"""CityFlow's first real Airflow DAG - three small tasks, wired with the
TaskFlow API, that read one real (tiny) CityFlow reference file, validate
what they read, and print a greeting."""
from datetime import datetime

import pandas as pd
from airflow.decorators import dag, task


@dag(
    dag_id="cityflow_hello",
    schedule=None,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["cityflow", "module-2"],
)
def cityflow_hello_dag():
    @task
    def extract_zone_count() -> int:
        """Read CityFlow's real NYC taxi-zone lookup table and return how
        many zones it defines."""
        zones = pd.read_csv("/opt/airflow/dags/data/taxi_zone_lookup.csv")
        return len(zones)

    @task
    def validate_zone_count(count: int) -> int:
        """A one-line sanity check: CityFlow always tracks at least one
        zone. Airflow passes `count` from the task above through XCom."""
        assert count > 0, f"expected at least one zone, got {count}"
        return count

    @task
    def greet_cityflow(count: int) -> None:
        print(f"Hello CityFlow! Tracking {count} real NYC taxi zones.")

    greet_cityflow(validate_zone_count(extract_zone_count()))


cityflow_hello_dag()

This block is marked # gate: skip because a DAG file isn’t meaningfully run by calling python cityflow_hello_dag.py in a bare virtual environment — from airflow.decorators import dag, task only resolves inside an actual Airflow install, and the whole point of a DAG is that Airflow’s scheduler runs it, not you. The real verification is below, against the live stack.

A few things worth naming about the code itself:

• @dag turns a plain Python function into a DAG definition. schedule=None means “only ever run when triggered” — no automatic cadence yet (Module 3 covers real schedules). catchup=False stops Airflow from trying to backfill runs for every interval since start_date, which would otherwise fire immediately on a start_date this far in the past.
• @task turns each inner function into one task. Calling extract_zone_count() inside the DAG function doesn’t run the function right now — it registers a task and returns a placeholder Airflow resolves later.
• greet_cityflow(validate_zone_count(extract_zone_count())) both wires the dependency chain and passes data: extract_zone_count’s return value flows into validate_zone_count as its argument, and that task’s return value flows into greet_cityflow. Airflow calls this mechanism XCom (cross-communication) — small values like a plain int pass through the metadata database between tasks automatically, which is exactly why this DAG’s dependency chain reads like ordinary nested function calls, not like the {task: [upstream_deps]} dictionaries Module 1 built by hand.

The file is saved to .airflow-local/dags/cityflow_hello_dag.py — the exact host folder docker-compose.yaml mounts to /opt/airflow/dags in both the webserver and scheduler containers.

The Scheduler Finds It#

Airflow’s scheduler scans the DAGs folder on its own, with no restart needed. Confirm it found the new file with the real CLI, against the real running scheduler container:

import subprocess
import time

deadline = time.time() + 60
dag_id = "cityflow_hello"
found = False

while time.time() < deadline:
    result = subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "list"],
        capture_output=True, text=True, check=True,
    )
    if dag_id in result.stdout:
        found = True
        break
    time.sleep(3)

print(result.stdout)
assert found, f"{dag_id} was not picked up by the scheduler within 60s"
print(f"scheduler picked up '{dag_id}'")

dag_id         | fileloc                                 | owners  | is_paused
===============+=========================================+=========+==========
cityflow_hello | /opt/airflow/dags/cityflow_hello_dag.py | airflow | None

scheduler picked up 'cityflow_hello'

No restart, no manual “reload DAGs” step — the file landed in the mounted folder and the scheduler’s own periodic scan found it, in this run within seconds.

Triggering It for Real#

New DAGs are paused by default (AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true', set back in Lesson 2’s compose file) — a deliberate safety default so a DAG with a real schedule can’t start firing runs before you’ve looked at it. Unpause it, then trigger a run with a unique run ID so this lesson’s verification can find exactly the run it started:

import subprocess
import time

dag_id = "cityflow_hello"
run_id = f"lesson3_verify_{int(time.time())}"

subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "unpause", dag_id],
    capture_output=True, text=True, check=True,
)
trigger = subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1",
     "airflow", "dags", "trigger", dag_id, "--run-id", run_id],
    capture_output=True, text=True, check=True,
)
print(f"triggered {dag_id!r} as run {run_id!r}")
print(trigger.stdout[-300:])

triggered 'cityflow_hello' as run 'lesson3_verify_1784738xxx'
     |                |                      |                      |                      |          |                  | last_scheduling_deci |                       |          |            |
conf | dag_id         | dag_run_id           | ...                  | ...                  | end_date | external_trigger | sion                 | logical_date          | run_type | start_date | state
=====+================+======================+======================+======================+==========+==================+======================+=======================+==========+============+=======
{}   | cityflow_hello | lesson3_verify_...   | 2026-07-22           | 2026-07-22           | None     | True             | None                 | 2026-07-22            | manual   | None       | queued

airflow dags trigger is the CLI equivalent of clicking the “play” button in the UI (Lesson 4 shows exactly where that button lives) — it creates a new DAG run row in the metadata database with state: queued, which the scheduler picks up on its very next cycle.

Polling to a Real Success#

Triggering starts a run; it doesn’t wait for it. Poll airflow dags list-runs — filtering by the exact run_id from the last step — until the run reaches a terminal state:

import json
import subprocess
import time

dag_id = "cityflow_hello"

# Re-derive the same run_id pattern this lesson just triggered, in case this
# block runs standalone: find the most recent run for this DAG instead.
result = 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(result.stdout)
latest = sorted(runs, key=lambda r: r["execution_date"])[-1]
run_id = latest["run_id"]

deadline = time.time() + 60
state = latest["state"]
while state not in ("success", "failed") and time.time() < deadline:
    time.sleep(3)
    result = 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(result.stdout)
    state = next(r["state"] for r in runs if r["run_id"] == run_id)

print(f"run {run_id!r} reached state: {state}")
assert state == "success", f"expected success, got {state}"

states = subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1",
     "airflow", "tasks", "states-for-dag-run", dag_id, run_id],
    capture_output=True, text=True, check=True,
)
print(states.stdout)

run 'lesson3_verify_1784738xxx' reached state: success
dag_id         | execution_date            | task_id             | state   | start_date                       | end_date
===============+===========================+=====================+=========+==================================+=================================
cityflow_hello | 2026-07-22T16:48:50+00:00 | extract_zone_count  | success | 2026-07-22T16:48:51.500977+00:00 | 2026-07-22T16:48:51.625352+00:00
cityflow_hello | 2026-07-22T16:48:50+00:00 | validate_zone_count | success | 2026-07-22T16:48:52.496616+00:00 | 2026-07-22T16:48:52.611412+00:00
cityflow_hello | 2026-07-22T16:48:50+00:00 | greet_cityflow      | success | 2026-07-22T16:48:53.558376+00:00 | 2026-07-22T16:48:53.657941+00:00

All three tasks, success, in under three seconds. And the actual greeting really printed — pulled straight from the task’s real log file inside the scheduler container:

[2026-07-22T16:48:53.653+0000] {logging_mixin.py:190} INFO - Hello CityFlow! Tracking 265 real NYC taxi zones.

265 — the same real zone count taxi_zone_lookup.csv has always had, now produced by a DAG the scheduler found, ran, and recorded success for entirely on its own, the exact gap Module 1 spent five lessons describing.

Key Takeaways#

• The TaskFlow API (@dag / @task) turns plain Python functions into a DAG; nesting calls like greet_cityflow(validate_zone_count(extract_zone_count())) both wires the dependency chain and passes data between tasks via XCom.
• A DAG-authoring file (from airflow.decorators import ...) is illustrative code, not something a bare Python process can run standalone — it’s only meaningfully executed by Airflow’s own scheduler, which is why this lesson verifies it with real CLI calls against the live stack instead.
• The scheduler finds a new .py file in the mounted dags/ folder on its own periodic scan — no restart required, confirmed here within seconds.
• airflow dags trigger starts a run; airflow dags list-runs / airflow tasks states-for-dag-run are how you check whether it actually succeeded — this lesson’s real run went queued → running → success for all three tasks in under three seconds.

Exercises#

Exercise 1 — Add a fourth task, count_greeting, that depends on greet_cityflow and simply prints how many times this DAG has now run (hint: airflow dags list-runs -d cityflow_hello --output json from a task is overkill — for this exercise, just hardcode a small counter and note in a comment that a real counter would need external state, which Module 4’s Hooks lesson addresses properly).

Exercise 2 — Trigger cityflow_hello twice in a row with two different --run-id values before either finishes. Use airflow dags list-runs -d cityflow_hello to confirm Airflow tracks both runs independently, each with its own task states.

Exercise 3 — Deliberately break validate_zone_count (e.g. assert count > 1_000_000) so it always fails, save the file, wait for the scheduler to notice the change, and trigger the DAG again. Use airflow tasks states-for-dag-run to confirm extract_zone_count still shows success while validate_zone_count and greet_cityflow show failed / upstream_failed — proof that a failure stops the chain rather than silently continuing.