Y

Apache Airflow and orchestration · Intermediate

A monthly pipeline as an Airflow DAG

4 min read

Why it matters

A script run by hand is not a pipeline. Teams need schedules, retries, history, alerts and a clear view of what ran and what failed. Airflow is the orchestrator most often named in data engineering job descriptions, and interviewers ask candidates to sketch a DAG for a pipeline they have built.

The concept

  • A DAG (directed acyclic graph) is a pipeline: tasks and the dependencies between them, defined in Python.

  • A task is one unit of work that can succeed, fail and retry on its own. Keep tasks small and meaningful: extract, validate, transform, load.

  • The schedule (a cron expression or preset) decides when runs are created. Each run has a logical date.

  • XCom passes small values (a path, a count) between tasks. Never pass data frames; pass references to where the data is stored.

  • The TaskFlow API (@dag, @task) turns Python functions into tasks and infers dependencies from how you call them.

Worked example: the TLC monthly load

The NYC TLC publishes each month's trips a few weeks later, so the job runs on the 3rd and retries every six hours if the file is not there yet. Each run processes the month before its run date.

dags/taxi_monthly.py
"""Monthly NYC taxi pipeline: extract, validate, transform. Each run processes the month before its run date."""
from datetime import datetime, timedelta
from pathlib import Path

import duckdb
from airflow.sdk import dag, task

SOURCE = "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_{month}.parquet"
LAKE = Path("/tmp/lake")


@dag(
    schedule="0 6 3 * *",                        # 06:00 UTC on the 3rd of each month
    start_date=datetime(2025, 2, 1),
    catchup=False,
    max_active_runs=1,
    default_args={"retries": 3, "retry_delay": timedelta(hours=6)},   # TLC often publishes late
    tags=["taxi"],
)
def taxi_monthly():
    @task
    def month_to_process(logical_date=None) -> str:
        first_of_month = logical_date.replace(day=1)
        return (first_of_month - timedelta(days=1)).strftime("%Y-%m")

    @task
    def extract(month: str) -> str:
        target = LAKE / "raw" / f"{month}.parquet"
        if not target.exists():                  # raw files are written once
            target.parent.mkdir(parents=True, exist_ok=True)
            duckdb.sql(f"copy (select * from '{SOURCE.format(month=month)}') to '{target}' (format parquet)")
        return str(target)

    @task
    def validate(raw: str, month: str) -> str:
        rows, outside = duckdb.sql(f"""
            select count(*), count(*) filter (where strftime(tpep_pickup_datetime, '%Y-%m') <> '{month}')
            from '{raw}'""").fetchone()
        if rows < 1_000_000 or outside / rows > 0.001:
            raise ValueError(f"{month}: {rows} rows, {outside} outside the month")
        return raw

    @task
    def transform(raw: str, month: str) -> int:
        out = LAKE / "clean" / f"month={month}" / "trips.parquet"
        out.parent.mkdir(parents=True, exist_ok=True)
        duckdb.sql(f"""
            copy (select tpep_pickup_datetime as pickup_at, PULocationID as pickup_zone,
                         DOLocationID as dropoff_zone, fare_amount, tip_amount
                  from '{raw}'
                  where strftime(tpep_pickup_datetime, '%Y-%m') = '{month}' and fare_amount >= 0)
            to '{out}' (format parquet)""")
        return duckdb.sql(f"select count(*) from '{out}'").fetchone()[0]

    month = month_to_process()
    transform(validate(extract(month), month), month)


taxi_monthly()

Test it without a scheduler or web server:

test-dag.sh
python -m venv .venv && source .venv/bin/activate
pip install "apache-airflow==3.1.*" duckdb
export AIRFLOW_HOME="$PWD" AIRFLOW__CORE__LOAD_EXAMPLES=False
airflow db migrate
airflow dags test taxi_monthly 2025-02-03

The run computes 2025-01, downloads the file, validates it and writes 3,331,086 cleaned trips; all four tasks end in success. Rerun it and the raw file is reused while the clean file is overwritten, so the result does not change.

Why compute the month from the logical date instead of using the data interval? In Airflow 3, cron schedules use a timetable whose runs have a zero-length data interval (start equals end), a change from Airflow 2. Deriving the period explicitly makes the DAG correct under either behaviour, and makes manual runs for any date do the right thing.

Common mistakes

  • Heavy work at the top level of the DAG file. The scheduler parses DAG files constantly; keep imports and code inside tasks.

  • Passing data through XCom. Pass paths or table names; store data in the lake or warehouse.

  • Using datetime.now() inside tasks. Use the run's logical date, so reruns and backfills process the right period.

  • One giant task. A failure then reruns everything. Split at natural checkpoints.

Interview angle

"Design the orchestration for a daily pipeline that depends on a partner's file." Describe the schedule, a sensor or retries for late data, small idempotent tasks, validation before load, alerts on failure, and how you would backfill. Draw the DAG.

Practice task

  1. Add a load_summary task that writes monthly trip counts per borough to a DuckDB table.

  2. Make validate fail on purpose (raise the row threshold) and watch the retries in the logs.

  3. Add on_failure_callback that writes a message to a file (standing in for a chat alert).

Further reading