Lab 04 — Run your first Airflow DAG

Lesson 04 · Data Engineering with Python · ~45 min

Goal: finish this lab with Airflow running locally, the gharchive_hourly DAG visible in the UI, and one successful manual run that produces the same hourly Parquet file Module 1 produced — but now triggered through Airflow.

Prereqs

• Module 1 labs completed — data-engineering/sample-pipeline/extract/ and transform/ have working implementations
• Docker Desktop (Mac/Windows) or docker compose v2+ (Linux). Run docker compose version to confirm
• 4 GB free RAM allocated to Docker Desktop (check: Docker Desktop → Settings → Resources)
• ~3 GB free disk for the Airflow image on first pull

Step 1 — Spin up Airflow

Open a terminal at the repository root:

cd data-engineering/sample-pipeline/airflow
docker compose up -d
docker compose logs -f airflow-init   # wait until you see "Initialization done"

The first run downloads the apache/airflow:2.9.3-python3.11 image (~1.5 GB) and the postgres:15 image (~400 MB). Subsequent starts are fast because Docker caches the images.

Once airflow-init exits successfully, open http://localhost:8080. Log in with username admin and password admin. You should see the Airflow home page with an empty DAGs list — no example DAGs, because AIRFLOW__CORE__LOAD_EXAMPLES: "false" is set in the compose file.

Step 2 — Verify the DAG was discovered

Airflow scans the dags/ folder every 30 seconds. Refresh the DAGs page after ~30-60 seconds. The gharchive_hourly DAG should appear with the tags data-engineering and module-2.

Click on gharchive_hourly to open it. You will see two views at the top:

• Grid view — empty for now (no runs yet). One row per task: extract and transform.
• Graph view — shows extract → transform as a two-node directed graph.

If the DAG does not appear after 60 seconds, check the "DAG Import Errors" banner at the top of the DAGs page. Any Python exception raised when Airflow imports the file will appear there.

Step 3 — Inspect the DAG file

The DAG file lives at data-engineering/sample-pipeline/airflow/dags/gharchive_hourly.py. Open it in your editor. Here is the full text for reference:

"""gharchive_hourly — Module 2 Lesson 04 sample DAG.

Wraps the existing extract + transform functions from Module 1 (Plan A)
into a two-task Airflow DAG. Lesson 04 deliberately uses schedule=None
(manual trigger only); Lesson 05 turns this into an hourly scheduled DAG.
"""

from __future__ import annotations

import sys
import pathlib
from datetime import datetime

from airflow import DAG
from airflow.operators.python import PythonOperator

# Make the sample-pipeline package importable from inside the Airflow container.
# /opt/airflow/sample-pipeline is the mount point set by docker-compose.yaml.
_SAMPLE_PIPELINE_ROOT = pathlib.Path("/opt/airflow/sample-pipeline")
if _SAMPLE_PIPELINE_ROOT.exists() and str(_SAMPLE_PIPELINE_ROOT) not in sys.path:
    sys.path.insert(0, str(_SAMPLE_PIPELINE_ROOT))


def _extract(**context) -> None:
    """L04: hardcoded hour. L05 will replace with context["data_interval_start"]."""
    from extract.gharchive import fetch_hour
    fetch_hour(hour="2026-05-01-12")


def _transform(**context) -> None:
    from transform.events import count_by_type
    count_by_type(
        input_path="data/raw/2026-05-01-12.json.gz",
        output_path="data/out/2026-05-01-12.csv",
    )


with DAG(
    dag_id="gharchive_hourly",
    start_date=datetime(2026, 5, 1),
    schedule=None,                       # L05 sets this to "@hourly"
    catchup=False,
    tags=["data-engineering", "module-2"],
    default_args={"owner": "data-eng"},
) as dag:
    extract = PythonOperator(task_id="extract", python_callable=_extract)
    transform = PythonOperator(task_id="transform", python_callable=_transform)
    extract >> transform

Walk through each section:

• Imports — DAG and PythonOperator come from Airflow; everything else is stdlib.
• sys.path injection — inside the Docker container, the sample-pipeline/ directory is mounted at /opt/airflow/sample-pipeline. We insert that path so Python can find extract.gharchive and transform.events. On your host machine this path doesn't exist, so the if _SAMPLE_PIPELINE_ROOT.exists() guard makes the file safe to parse outside Docker.
• Deferred imports — from extract.gharchive import fetch_hour is inside the function body, not at module scope. This keeps parse time fast (Airflow parses this file every 30s).
• DAG context manager — the with DAG(...) as dag: block is the standard modern Airflow pattern. All operators defined inside this block are automatically registered to this DAG.
• Task dependency — extract >> transform declares that transform should only run after extract succeeds.

Step 4 — Trigger a manual run

In the Airflow UI, navigate to the gharchive_hourly DAG:

1. Click the Play button (triangle icon, top-right of the DAG page)
2. Select Trigger DAG
3. Leave the JSON config blank and click Trigger

Switch to Grid view. Watch the cells change colour: light yellow (queued) → yellow (running) → dark green (success). The extract task runs first; once it turns green, transform starts.

Click the green extract cell and then click Log. You should see the download messages from gharchive.py: something like Downloading 2026-05-01-12.json.gz... followed by a byte count.

Step 5 — Verify the output

The DAG writes its output to data-engineering/sample-pipeline/data/ on your host (because the ../ volume mount makes your host filesystem visible inside the container at /opt/airflow/sample-pipeline). Open a terminal and check:

ls data-engineering/sample-pipeline/data/
# Expected: 2026-05-01-12.parquet (or similar)

Query the output with DuckDB:

duckdb -c "SELECT COUNT(*), event_type FROM read_parquet('data-engineering/sample-pipeline/data/2026-05-01-12.parquet') GROUP BY event_type ORDER BY count DESC LIMIT 5"

The row counts should match what you got when you ran the Module 1 script directly. Airflow did not change what was computed — it changed how it was triggered and monitored.

Step 6 — Tour the UI

While the DAG is fresh, explore the UI panels:

• Grid view — one cell per (task, run) pair. Hover over the cell for the duration and start time. Click for full details.
• Graph view — the DAG topology. Click a node to see its task instance detail.
• Code — what Airflow actually parsed. Confirm it matches the file on disk.
• Task Logs — click any task cell → "Log" to stream the full stdout/stderr from that attempt.
• Trigger w/ config — click Play → Trigger DAG w/ config. Leave the JSON box blank for now; Lesson 05 uses it to pass the data interval.

Step 7 — Trigger a failure on purpose

Open data-engineering/sample-pipeline/airflow/dags/gharchive_hourly.py in your editor. Change the hardcoded hour to "2099-01-01-00" — that date is in the far future, so gharchive.org will return a 404.

def _extract(**context) -> None:
    from extract.gharchive import fetch_hour
    fetch_hour(hour="2099-01-01-00")  # intentional 404

Save the file. Wait ~30 seconds for Airflow to re-parse. Then trigger another manual run. The extract task will turn red (failed). The transform task will turn light grey (upstream failed — skipped). Open the extract logs to see the HTTP 404 error.

Before moving on, revert this change — restore the hour to "2026-05-01-12".

What you should leave with

• A running Airflow stack accessible at http://localhost:8080
• One successful gharchive_hourly DagRun visible in the Grid view
• Familiarity with: Grid view, Graph view, Task Logs, Code, and manual Trigger
• Experience watching a task fail and reading the failure logs

Troubleshooting

python -m py_compile data-engineering/sample-pipeline/airflow/dags/gharchive_hourly.py