Week 20: Data Pipelines & Batch Processing with Airflow

1. DAGs, Tasks & the Scheduler Model

An Airflow DAG (Directed Acyclic Graph) is a Python file that declares a set of tasks and the dependencies between them — which tasks must finish before others can start. "Acyclic" is a real constraint, not just terminology: a DAG can never loop back on itself, because Airflow needs to compute a definite execution order, and a cycle would mean two tasks each waiting on the other forever.

from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(
    schedule="0 2 * * *",   # 02:00 UTC daily
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def daily_orders_report():

    @task
    def extract_orders() -> str:
        # pulls yesterday's orders from the production database
        return export_orders_to_parquet()

    @task
    def transform_orders(raw_path: str) -> str:
        return aggregate_daily_totals(raw_path)

    @task
    def load_to_warehouse(aggregated_path: str) -> None:
        load_into_warehouse_table(aggregated_path)

    load_to_warehouse(transform_orders(extract_orders()))

daily_orders_report()

Calling transform_orders(extract_orders()) does two things at once: it declares the dependency (transform_orders can't run until extract_orders finishes) and wires extract_orders's return value into transform_orders's argument, via Airflow's XCom mechanism — a small key-value store the scheduler uses to pass data between tasks. This is the same explicit-dependency idea as Week 18's saga steps, but for a scheduled batch pipeline instead of an event-driven one: each task only needs to know what it depends on, and the scheduler assembles the full execution order from those declarations.

2. A Real ETL Pipeline

Fleshing out Section 1's skeleton with real logic: Extract pulls data from a source system into a raw, intermediate form; Transform cleans, validates, and aggregates it; Load writes the result to its final destination — a pattern that generalizes to almost any pipeline regardless of the specific data involved.

import pandas as pd
from sqlalchemy import create_engine

def export_orders_to_parquet() -> str:
    engine = create_engine(PROD_DB_URL)
    df = pd.read_sql(
        "SELECT * FROM orders WHERE created_at >= now() - interval '1 day'",
        engine,
    )
    path = f"/data/raw/orders_{datetime.now():%Y%m%d}.parquet"
    df.to_parquet(path)
    return path

def aggregate_daily_totals(raw_path: str) -> str:
    df = pd.read_parquet(raw_path)
    daily = df.groupby(df["created_at"].dt.date).agg(
        total_orders=("id", "count"),
        total_revenue=("amount", "sum"),
    )
    out_path = raw_path.replace("raw", "aggregated")
    daily.to_parquet(out_path)
    return out_path

def load_into_warehouse_table(aggregated_path: str) -> None:
    df = pd.read_parquet(aggregated_path)
    engine = create_engine(WAREHOUSE_DB_URL)
    df.to_sql("daily_order_summary", engine, if_exists="append", index=True)

Notice the production database (PROD_DB_URL) and the warehouse (WAREHOUSE_DB_URL) are deliberately separate — the extract step reads from the same database your FastAPI service writes to, but the aggregation and load steps run against a warehouse built for analytical queries, so a heavy daily aggregation never competes with your API's live traffic for the same database's resources.

3. Retries, Backfills & Monitoring a Production DAG

default_args={"retries": 2, "retry_delay": timedelta(minutes=5)} from Section 1 means a failed task — a transient network blip pulling from the source database, say — is retried automatically before the whole DAG is marked failed, the same instinct behind Week 9's retry patterns for HTTP calls, applied to a scheduled pipeline instead of a live request.

A backfill runs a DAG for a range of past dates it either never ran for, or needs to rerun after a bug fix:

airflow dags backfill daily_orders_report \
  --start-date 2026-07-01 \
  --end-date 2026-07-14

This is exactly why every task needs to be idempotent per the previous callout — a backfill reruns each day's tasks, potentially including days that already succeeded once before, and it should be safe to do so without producing duplicate or inconsistent data in the warehouse.

The Airflow web UI's Grid view shows every DAG run and every task's status (success, failed, retried, running) at a glance, which is the first place to look when a pipeline needs attention — but for genuine production monitoring, wire Airflow's task failure callbacks into the same alerting channel from Week 12's observability practices, rather than relying on someone remembering to check the UI:

def notify_failure(context):
    task_id = context["task_instance"].task_id
    dag_id = context["dag"].dag_id
    send_alert(f"Airflow task failed: {dag_id}.{task_id}")

@dag(
    schedule="0 2 * * *",
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "on_failure_callback": notify_failure,
    },
)
def daily_orders_report():
    ...

4. Hands-on Exercise

5. Knowledge Check

Three quick questions. Expand each to check your answer.