Orchestration with Apache Airflow

What you will build: an Airflow pipeline (a DAG) that runs every day, extracts orders, calculates revenue and saves a CSV report, with automatic retries and a clear view of what succeeded or failed.

Prerequisites: Python, the command line, and the ideas from *ETL vs ELT and pipeline design*. Apache Airflow does not run natively on Windows. Windows users should use WSL 2 (Ubuntu) or Docker; macOS and Linux work directly. This lesson is written for Airflow 3 (the current major version at the time of writing, 2026); notes mark what differs in Airflow 2.

A pipeline is rarely one script. It is several steps that must run in order, at the right time, and recover when something fails: extract from an API, load to the warehouse, run dbt, refresh a dashboard. You could glue these together with cron, but cron cannot tell you that step 2 failed, retry just that step, wait for step 1 to finish, or re-run last Tuesday.

Apache Airflow is an open-source orchestrator. You describe the workflow in Python; Airflow schedules it, runs the steps in the right order, retries failures, keeps a history and shows everything in a web UI. Airflow does not process big data itself. It tells other tools (Spark, dbt, a SQL warehouse, a Python script) when to run. Think of it as the shift manager, not the cook.

A DAG is "acyclic" so that there is always a clear order: a task can depend on earlier tasks but never on itself through a loop. A DAG run covers a data interval (for example one day), and by default Airflow starts the run for an interval after that interval has ended. A @daily run covering 3 October starts just after midnight on 4 October, because only then is all of 3 October's data available.

airflow standalone is for learning on one machine. Production uses a proper database (PostgreSQL), separate services, and often Docker Compose, Kubernetes or a managed service.

Airflow reads Python files from the dags folder inside AIRFLOW_HOME (so airflow-demo/airflow/dags). Create the folder if it does not exist. Airflow's TaskFlow API turns plain Python functions into tasks with the @task decorator and wires dependencies from how you call them.

import csv
import json
from datetime import datetime, timedelta
from pathlib import Path

from airflow.sdk import dag, task   # Airflow 3. On Airflow 2: from airflow.decorators import dag, task

OUT_DIR = Path.home() / "brew_reports"

MENU = {
    "espresso": 120, "cappuccino": 180, "cold-brew": 220, "filter-coffee": 80,
    "masala-chai": 90, "croissant": 150, "banana-bread": 130, "veg-sandwich": 160,
}


@dag(
    dag_id="brew_daily_sales",
    schedule="@daily",
    start_date=datetime(2026, 10, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
    tags=["brew-and-bean"],
)
def brew_daily_sales():

    @task
    def extract(ds=None):
        """Pretend to fetch yesterday's orders. In real life: call the API or read the DB."""
        print(f"Extracting orders for {ds}")
        return [
            {"order_id": 1, "item_id": "cappuccino", "qty": 2},
            {"order_id": 2, "item_id": "espresso", "qty": 1},
            {"order_id": 3, "item_id": "croissant", "qty": 2},
        ]

    @task
    def transform(rows):
        total = 0
        for row in rows:
            if row["item_id"] not in MENU:
                raise ValueError(f"Unknown item_id: {row['item_id']}")
            row["revenue"] = row["qty"] * MENU[row["item_id"]]
            total += row["revenue"]
        return {"rows": rows, "total": total}

    @task
    def load(result, ds=None):
        OUT_DIR.mkdir(parents=True, exist_ok=True)
        path = OUT_DIR / f"sales_{ds}.csv"
        with open(path, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=["order_id", "item_id", "qty", "revenue"])
            writer.writeheader()
            writer.writerows(result["rows"])
        print(f"Wrote {path}; day total = {result['total']}")

    load(transform(extract()))


brew_daily_sales()

A three-task DAG. extract returns a list; transform checks and prices it; load writes sales_<date>.csv to a brew_reports folder in your home directory. The ds argument is the run's date as YYYY-MM-DD, which Airflow fills in.

Walk through what each part does:

@dag(...) defines the DAG. schedule="@daily" means once a day. start_date is the first date Airflow considers. catchup=False stops Airflow from creating one run for every day since the start date when you first switch the DAG on (with catchup=True it would backfill them all, which is useful for history and surprising when unexpected).

default_args applies to every task. Here each task retries twice, waiting five minutes between tries, which handles brief network failures without waking anyone.

Each @task function becomes a task. The last line, load(transform(extract())), says: run extract, pass its result to transform, pass that to load. Airflow draws the dependency graph from this.

In transform, an unknown item_id raises an error. That makes the task fail loudly, which is what you want: bad data should stop the pipeline, not sneak into reports.

The total is 2 x 180 + 120 + 2 x 150 = 780. Airflow also gives you airflow dags list to confirm it found your file, and airflow dags list-import-errors to see why a DAG file failed to load (a typo shows up there).

Failures and re-runs are normal in orchestration, so design tasks to be idempotent: running the same task twice for the same date gives the same result as running it once. Our load task is idempotent because the file name contains the date and the file is overwritten. A task that blindly appends rows to a table would create duplicates on a retry. Use "delete that day's rows then insert", MERGE/upsert, or overwrite a date partition instead.

Always base work on the run's date (ds, or the data interval) and never on datetime.now(). Then a backfill for last month processes last month's data, not today's.

Typical production use: Airflow triggers an ingestion job, then a Spark or dbt run (see *Apache Spark and PySpark basics* and *Transformations with dbt*), then data quality checks (see *Data quality, testing and observability*), then a notification. Alternatives you may meet are Dagster, Prefect, and the schedulers built into cloud platforms (for example Databricks Workflows or cloud-managed Airflow services). The ideas (DAG, retries, idempotency, backfills) transfer to all of them, which is why interviewers ask about them.

How this comes up in interviews: "What is a DAG?", "What is the difference between catchup and backfill?", "How do you make a task idempotent?", "How do tasks share data?" (XCom for small values, external storage for large), and "Why not just use cron?".

You are a senior data engineer. Review this Airflow DAG for production readiness. Check: idempotency, use of the run date instead of now(), retries, XCom size, secrets in code, and clarity of task boundaries. List problems in order of importance with a fixed code snippet for each.

[paste your DAG file]

What to expect: A good review points out hard-coded passwords, non-idempotent writes, use of datetime.now(), large XCom payloads and tasks that do too much. Check each suggestion before applying it.