Airflow in Production: The Full Pipeline Case Study

Thirty-six articles of incidents, and they all rhyme. The cron job that double-billed. The sensor that burned a worker slot. The /tmp that made reruns nondeterministic. The schema change that skipped review. Every one of those failures was preventable with the same machinery — this capstone builds all of it at once.

The setup: a mid-size retailer, 60M order rows a day, a Snowflake warehouse, and a mandate to make analytics trustworthy by morning. Two engineers, six weeks, one Airflow platform.

This is the full case study: the design doc first, then the build in phases — idempotent ETL, quality gates, alerting, CI/CD, and the incident runbook that holds it together.

The toy DAG era ends here.

If you take one thing from the whole series, let it be this: production is a system, not a script. Design it, test it, monitor it, and write down what to do when it breaks — you'll thank yourself at 2 AM.

1. The Company, the Problem, the Constraints

The company is a mid-size retailer with 60M order rows a day: web orders, store POS, returns, refunds. Analytics ran on cron jobs and ad-hoc scripts — the cron job that double-billed every customer from the series opener was not a metaphor; it was this company, three years earlier. The mandate: trustworthy analytics by 6 AM, every morning.

The constraints are what made it a real build. No downtime window — the legacy pipelines keep running until the new ones prove themselves. Two engineers, six weeks. One warehouse (Snowflake) and a dozen sources with varying reliability. And a data team that had been burned enough to demand evidence: row counts, freshness checks, and a paper trail.

That last constraint is the whole story. Reliability wasn't a tool problem; it was a system problem. The build plan was written as a design doc before the first DAG existed.

2. The Design Doc: DAGs, Schedules, Contracts

The design doc was one page of tables, not an essay. Three pipeline families: orders (core, 3 AM, SLA 5:30), returns (core, 4 AM, SLA 6:00), and partner sync (standard, 2 AM, best-effort). Each family had an owner, a freshness SLA, and a contract: schema, natural key, and what the consumers could expect at what time.

The contracts were the part the old platform never had. A data contract is a promise: orders_ods has exactly one row per order_id, is complete for the interval, and is available by the SLA time. Writing it down first turned every later quality gate into a test of the contract instead of an opinion.

DAG design followed the contracts: one DAG per family, idempotent tasks, max_active_runs=1, catchup=False. The doc also recorded the failure playbook pointers — which article of this series applied to which failure mode. The doc was the runbook's table of contents.

# dag_manifest.yaml — the platform contract
pipelines:
  market_etl:
    schedule: "0 3 * * *"
    family: core
    sla: "complete by 05:30 UTC"
    natural_key: order_id
    quality_gates: [row_counts, dedupe, freshness]
    owners: [data-platform]
  returns_etl:
    schedule: "0 4 * * *"
    family: core
    sla: "complete by 06:00 UTC"
    natural_key: return_id
    quality_gates: [row_counts, dedupe]
    owners: [data-platform]
  partner_sync:
    schedule: "0 2 * * *"
    family: standard
    sla: best_effort
    quality_gates: [row_counts]
    owners: [integrations]

3. Build Phase 1: Idempotent ETL

Phase one shipped the core DAG: ingest orders, transform, load. Three tasks, every one idempotent. Ingest downloaded to object storage keyed by data_interval_end — the /tmp incident from this series was the reason for that rule. Transform read the staged object and wrote clean objects. Load was a MERGE keyed on order_id.

Idempotency was non-negotiable from day one: the rerun-that-loaded-twice incident was the failure this company could not repeat. Every load task was written so that clearing and rerunning produced exactly the same state as the original run. The proof was part of the definition of done: clear the interval, rerun, diff the row counts — they must match.

The small details carried the weight. max_active_runs=1 so overlapping runs never raced the same partition. retries=3 with exponential backoff, per the queued-task lesson. catchup=False, with backfills only through the explicit command from the runbook. None of these were defaults; all of them were decisions.

Retries have a documented precedence the team leaned on: task-level retries, then default_args, then the deployment-wide AIRFLOW__CORE__DEFAULT_TASK_RETRIES env var — set the floor once at the platform level, override per task. The docs' incremental rule also matches this build: each run processes only its data interval's records (filter by last-modified date or sequence IDs), so a failure in one interval never blocks or re-touches the others.

from datetime import datetime, timedelta

from airflow import DAG
from airflow.decorators import task
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

with DAG(
    dag_id="market_etl",
    schedule="0 3 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=["core", "sla_critical"],
) as dag:

    @task(retries=3, retry_delay=timedelta(minutes=5), max_retry_delay=timedelta(hours=1))
    def ingest_orders(interval_end: str) -> str:
        src = ObjectStoragePath(f"s3://raw/orders/{interval_end}/feed.json")
        return src.as_uri()

    @task
    def transform_orders(src_uri: str) -> str:
        raw = ObjectStoragePath(src_uri)
        clean = ObjectStoragePath(f"s3://clean/orders/{src_uri.split('/')[-2]}/orders.parquet")
        clean.parent.mkdir(parents=True, exist_ok=True)
        clean.write_bytes(normalize_orders(raw.read_bytes()))
        return clean.as_uri()

    load_orders = SnowflakeOperator(
        task_id="load_orders",
        snowflake_conn_id="snowflake_prod",
        sql="""
        MERGE INTO orders_ods t
        USING (SELECT $1::varchar AS order_id, $2::number AS amount
               FROM @staging/orders/{{ data_interval_end }}/orders.parquet) s
        ON t.order_id = s.order_id
        WHEN MATCHED THEN UPDATE SET amount = s.amount, loaded_at = CURRENT_TIMESTAMP()
        WHEN NOT MATCHED THEN INSERT (order_id, amount, loaded_at)
        VALUES (s.order_id, s.amount, CURRENT_TIMESTAMP())
        """,
    )

    load_orders << transform_orders(ingest_orders("{{ data_interval_end }}"))

4. Build Phase 2: Quality Gates and Monitoring Wiring

Phase two added the gates that turned the contract into tests. After load, three check tasks: row counts within a band of the expected volume, zero duplicate natural keys, and freshness — the interval's max loaded_at within the SLA window. A gate failure fails the DAG and pages, because silent data bugs are how trust dies.

Monitoring was wired to the three primaries from the monitoring article: scheduler heartbeat, queued-task age, and SLA misses. The dead-scheduler-nine-days incident was the cautionary tale; this platform alerts on scheduler health within five minutes of it dying. Queue-age alerts catch the sensor-slot-waste pattern from the deferrable article before it costs a window.

The alerting philosophy: page on three things, log everything else. The SLA miss alert carried the contract name and the owner. Everything else — retries, warnings, backfills — went to a channel nobody had to watch.

Logs are data too. On disposable or autoscaled nodes, task logs die with the node, so the production-deployment docs point logs at a distributed filesystem (S3, GCS) or an external service (Elasticsearch, CloudWatch, Stackdriver) — this platform shipped remote logging in the same sprint as the alerts.

from airflow import AirflowException
from airflow.decorators import task


@task

def check_row_counts(expected: int) -> None:
    rows = run_query("SELECT count(*) FROM orders_ods WHERE loaded_date = CURRENT_DATE")
    if rows < expected * 0.99:
        raise AirflowException(f"row count {rows} below 99% of expected {expected}")


@task

def check_dedupe() -> None:
    dups = run_query(
        "SELECT count(*) FROM (SELECT order_id FROM orders_ods "
        "GROUP BY order_id HAVING count(*) > 1)"
    )
    if dups:
        raise AirflowException(f"{dups} duplicate order_ids in orders_ods")


@task

def check_freshness(sla: str) -> None:
    latest = run_query("SELECT max(loaded_at) FROM orders_ods")
    if latest > sla:
        raise AirflowException(f"max loaded_at {latest} exceeds SLA {sla}")

5. CI/CD and Rollout

Phase three made deploys boring. Every merge to main runs the pipeline from the testing article: lint, parse check via DagBag, DAG-load tests, and unit tests on the callables. The git-sync-deployed-a-broken-DAG incident was the cautionary tale — this platform deploys images built from the tested commit, never a file copy.

The gate is the point: a DAG that fails to parse is a merge-time failure, not a 2 AM outage. CI runs airflow dags list-import-errors against the image, and the build fails on any import error before anything reaches the scheduler. Rollout follows branch promotion: feature DAGs land behind tags, and core DAGs promote through staging with a canary run before prod.

Rollback is a previous image, not a mystery. Every deploy is a versioned image; reverting is redeploying the last green build. The registry keeps the last 20 so 'what was running when this data went bad' is always answerable.

Under the hood this is Airflow 3's dag-bundle mechanism: the scheduler sends instructions — 'run task X of DAG Y' — never the files, so every node must carry the bundle. GitDagBundle versioning is the docs' production pattern for that, and it's the same guarantee this team got from image-built bundles. Patch-level upgrades stay zero-downtime when the metadata schema is unchanged — test the upgrade in staging before the first live one.

# .github/workflows/dags.yml — the merge gate
name: dags
on:
  pull_request:
    paths: ["dags/**", "tests/**"]
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt
      - run: python -m pytest tests/ -x                # DagBag + unit tests
      - run: airflow dags list-import-errors --output table   # parse gate
      - run: docker build -t airflow-dags:${{ github.sha }} .
      - run: docker push registry.thecodeforge.io/airflow-dags:${{ github.sha }}

6. The Incident Runbook

The runbook shipped in the same sprint as the alerting — the 9-day silent outage from the monitoring article was the reason. It's organized by symptom, not by guess: each entry has the first commands, the likely cause, and the escalation path. The quick reference at the top of this article is its heart.

The runbook is the series playbook in one place. Scheduler heartbeat down → check scheduler health and DB. Quality gate failed → read the gate log, fix the source, clear and rerun. Import errors after deploy → run the CI gate locally, fix, redeploy — never a manual copy. Duplicate rows after rerun → verify the upsert key. Backfill storm → catchup=False, targeted backfills only.

Every incident writes back into the runbook. The post-mortem ends with a runbook diff: what entry was missing, what command was wrong, what escalation was too slow. The runbook is a living artifact precisely because the first version is always incomplete.

# Runbook — market_etl

## Quality gate failed
1. Open the gate task log:
   `airflow tasks test market_etl quality_gate <date>`
2. Find the threshold breach (row count / dedupe / freshness).
3. Fix the source or the contract — never the gate to pass.
4. Clear the interval and rerun:
   `airflow tasks clear market_etl --start-date <d> --end-date <d>`

## Scheduler heartbeat alert
1. `docker compose ps scheduler`
2. `curl -s http://localhost:8793/health`
3. Restart scheduler; verify DB connectivity.

## Escalation
- 1st: on-call data engineer (5 min)
- 2nd: data platform lead (15 min)
- 3rd: warehouse DBA (30 min)

7. From Mid-Level to Senior: The Roadmap

The senior roadmap is visible in everything this build did differently. Juniors write tasks; seniors write contracts and gates. Juniors fix the failing DAG; seniors ask which guarantee was violated and why the check didn't catch it earlier. Juniors monitor the logs; seniors monitor the three primaries and own the SLA.

The concrete moves: own a platform, not a DAG. Write the design doc and update it. Size the platform — executors, pools, parallelism — and defend the numbers. Automate the recovery so the runbook is rarely needed, then drill it anyway. Mentor the next person on the same playbook: idempotency, quality gates, alerting, CI/CD, runbook.

The interview answers from this series are the same discipline. When a senior question asks about reruns, you answer idempotency and upsert keys. When it asks about failures, you answer heartbeats and SLA misses. The series taught the incidents; the roadmap is making them your default thinking.

8. Course Wrap-Up: The Playbook, Complete

This closes the series, so the wrap-up is the whole playbook in one breath. Foundations: DAGs are code — test them, keep top-level code pure, and let the scheduler do scheduling. Authoring: connections hold credentials, XComs carry references, sensors defer. Scheduling: data intervals, catchup discipline, backfills as explicit operations.

Deployment and ops: pick the executor for the scale, operate scheduler, workers, and triggerer like infrastructure, monitor the three primaries, and tune parsing before buying machines. Security and quality: least privilege, secret backends, idempotent loads, gates on contracts. Advanced: deferrable operators, object storage, HITL — the Airflow 3 depth that made the platform genuinely modern.

The capstone built the system that every incident in the series was pointing at. The cron job that double-billed became the DAG with states and retries. The sensor that burned a slot became the deferrable wait. The /tmp that broke reruns became deterministic object keys. The schema change that skipped review became the HITL gate.

Now go build it. The incidents are the syllabus, the playbook is the graduation — and the warehouse is waiting for its 6 AM data.

One more docs rule to carry forward: XCom is for small messages, remote storage for large payloads — push the S3 path, not the data — and when XCom traffic gets heavy, a custom XCom backend keeps the metadata DB from becoming the bottleneck.