A cron job runs your script at 2 a.m. and says nothing when it fails; you find out at 9, staring at a dashboard full of holes. This Airflow tutorial replaces that fragile setup with a real data pipeline: one that retries network hiccups, refuses to load bad data, and can rerun any day without making a mess, built end to end on real data.

“A pipeline is not code that runs once. It is code that has to run every day, survive a bad night, and still be correct in the morning.”

Common wisdom among data engineers

Last Updated: July 2026 | Tested on: Python 3.14.6, Apache Airflow 3.x, pandas 2.3.3, pandera 0.32.1 | Difficulty: Intermediate | Reading Time: 18 minutes

Why cron Breaks: The DAG Mental Model

Picture a small tiffin kitchen that promises fresh dabbas by noon. The cook has a checklist: buy vegetables, wash them, cook, pack, then hand the boxes to the delivery rider. If the vegetable van is late, the cook waits and tries again, they do not start cooking air. If the vegetables look rotten, the whole batch stops right there, nobody packs spoiled food. A plain cron job is a cook with a stopwatch and no checklist: it starts each step at a fixed time whether or not the previous step finished, never retries, and never checks quality.

That is exactly where cron falls apart for data work. It has no idea that “transform” depends on “extract” finishing first. It cannot retry a step that failed because the Application Programming Interface (API) blinked. And if you need to rerun last Tuesday because the source data was fixed, cron has no memory of Tuesday at all. Airflow replaces the stopwatch with a DAG, a Directed Acyclic Graph. That is a fancy name for a simple idea: a list of tasks with arrows showing what must finish before what, and no arrow ever loops back on itself.

The diagram below is the exact pipeline we are about to build, including the retry loop on extract and the two paths where a task can fail and stop everything downstream.

Read the arrows as “must happen before”. The scheduler fires on the interval, extract runs and may retry a few times if the API times out, and only when clean rows come back does validate run. A bad row sends the run to the failure path so nothing downstream touches it. When every task passes, notify runs last. That dependency graph, plus automatic retries and a full history of every run, is the whole reason Airflow exists.

Airflow Tutorial Setup: Install and Run Standalone

Airflow targets Linux and macOS. On Windows, run it inside WSL2, which behaves like Linux. The fastest way to see it working is airflow standalone: one command that starts the scheduler, the API server, and the web UI together, and prints a login for you. Install with the official constraints file so the dozens of dependencies resolve to versions the Airflow team actually tested.

📄 terminal: install Airflow and start it

export AIRFLOW_HOME=~/airflow
AIRFLOW_VERSION=3.0.6
PY=$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)
CONSTRAINTS="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PY}.txt"
pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINTS}"
pip install pandas pandera requests
airflow standalone

▶ Example output (from WSL2, trimmed)

standalone | Starting Airflow Standalone
standalone | Checking database is initialized
scheduler  | Starting the scheduler
apiserver  | Application startup complete. Uvicorn running on http://0.0.0.0:8080
standalone | Airflow is ready
standalone | Login with username: admin  password: 8Xn2kQpR7mVw

What happened here: the install pulls Apache Airflow 3.x (the version named in most current job postings at the time of writing) with a pinned constraints file so it does not fight your other packages. airflow standalone then boots everything and hands you a password. Open http://localhost:8080, log in, and you get the grid view where each DAG run is a column and each task is a colored square: green for success, red for failed, yellow for up-for-retry.

That grid is the single best reason to use Airflow over a pile of scripts, because you can see at a glance which night broke and click straight into that task’s logs. Airflow discovers DAGs by scanning the dags/ folder inside AIRFLOW_HOME, so the file we write next just needs to land there.

Your First DAG: Extract, Validate, Transform, Load, Notify

Here is the heart of this Airflow tutorial: the whole pipeline as a DAG using the TaskFlow API, where each Python function decorated with @task becomes a node in the graph and the return value of one task flows into the next. Say a data engineer named Aditi needs daily currency rates landed in a database. The DAG pulls rates from a public API, runs them through a schema gate, reshapes them, writes them to SQLite, and sends a summary. Notice how the dependency graph at the bottom is just plain Python function calls: Airflow reads those calls to build the arrows in the grid.

📄 dags/fx_pipeline.py: the full DAG (TaskFlow API)

import pendulum
from airflow.sdk import dag, task   # Airflow 3.x authoring API

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 7, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=2)},
    tags=["etl", "currency"],
)
def fx_pipeline():

    @task
    def extract() -> dict:
        import requests
        api = "https://api.frankfurter.app/latest?from=USD&to=EUR,GBP,INR,JPY"
        payload = requests.get(api, timeout=10).json()
        rows = [{"currency": c, "rate": r} for c, r in payload["rates"].items()]
        return {"as_of": payload["date"], "rows": rows}

    @task
    def validate(raw: dict) -> dict:
        import pandas as pd, pandera.pandas as pa
        from pandera.pandas import Column, DataFrameSchema, Check
        schema = DataFrameSchema({
            "currency": Column(str, Check.str_length(3, 3)),
            "rate": Column(float, Check.greater_than(0)),
        })
        df = schema.validate(pd.DataFrame(raw["rows"]))  # raises on bad data
        return {"as_of": raw["as_of"], "rows": df.to_dict("records")}

    @task
    def transform(clean: dict) -> dict:
        import pandas as pd
        df = pd.DataFrame(clean["rows"])
        df["usd_per_unit"] = (1 / df["rate"]).round(4)
        df["as_of"] = clean["as_of"]
        return {"as_of": clean["as_of"], "rows": df.to_dict("records")}

    @task
    def load(shaped: dict) -> int:
        import sqlite3, pandas as pd
        df = pd.DataFrame(shaped["rows"])
        con = sqlite3.connect("/tmp/rates.db")
        con.execute("""CREATE TABLE IF NOT EXISTS fx_rates(
            as_of TEXT, currency TEXT, rate REAL, usd_per_unit REAL,
            PRIMARY KEY (as_of, currency))""")
        con.execute("DELETE FROM fx_rates WHERE as_of = ?", (shaped["as_of"],))
        df.to_sql("fx_rates", con, if_exists="append", index=False)
        con.commit()
        total = con.execute("SELECT COUNT(*) FROM fx_rates").fetchone()[0]
        con.close()
        return total

    @task
    def notify(total: int):
        print(f"Pipeline OK, table now holds {total} rows")

    # dependency graph: each call wires one task to the next
    notify(load(transform(validate(extract()))))

fx_pipeline()

You do not have to spin up the scheduler to test a DAG. The command airflow dags test fx_pipeline runs the whole graph once, in order, right in your terminal, using real task code.

▶ Example output: airflow dags test fx_pipeline (trimmed)

[2026-07-09] INFO - Marking task as SUCCESS. dag_id=fx_pipeline, task_id=extract
[2026-07-09] INFO - Marking task as SUCCESS. dag_id=fx_pipeline, task_id=validate
[2026-07-09] INFO - Marking task as SUCCESS. dag_id=fx_pipeline, task_id=transform
[2026-07-09] INFO - Marking task as SUCCESS. dag_id=fx_pipeline, task_id=load
[2026-07-09] INFO - notify - Pipeline OK, table now holds 4 rows
[2026-07-09] INFO - Marking task as SUCCESS. dag_id=fx_pipeline, task_id=notify
[2026-07-09] INFO - Dagrun Running -> success. dag_id=fx_pipeline

Since Airflow itself does not run on native Windows, let us prove the logic that lives inside those five tasks is correct by running the exact same functions as a plain Python script. This is the real captured output on Python 3.14.6, hitting the live API, twice in a row.

📄 pipeline_core.py: the same five steps, run without Airflow

import sqlite3, requests
import pandas as pd
import pandera.pandas as pa
from pandera.pandas import Column, DataFrameSchema, Check

API = "https://api.frankfurter.app/latest?from=USD&to=EUR,GBP,INR,JPY"
schema = DataFrameSchema({
    "currency": Column(str, Check.str_length(3, 3)),
    "rate": Column(float, Check.greater_than(0)),
})

def extract():
    p = requests.get(API, timeout=10).json()
    rows = [{"currency": c, "rate": r} for c, r in p["rates"].items()]
    return {"as_of": p["date"], "rows": rows}

def validate(raw):
    df = schema.validate(pd.DataFrame(raw["rows"]))   # gate: raises on bad data
    df["as_of"] = raw["as_of"]
    return df

def transform(df):
    df = df.copy()
    df["usd_per_unit"] = (1 / df["rate"]).round(4)
    return df.sort_values("currency").reset_index(drop=True)

def load(df):
    con = sqlite3.connect("rates.db")
    con.execute("""CREATE TABLE IF NOT EXISTS fx_rates(
        as_of TEXT, currency TEXT, rate REAL, usd_per_unit REAL,
        PRIMARY KEY (as_of, currency))""")
    run_date = df["as_of"].iloc[0]
    con.execute("DELETE FROM fx_rates WHERE as_of = ?", (run_date,))  # idempotent
    df[["as_of","currency","rate","usd_per_unit"]].to_sql(
        "fx_rates", con, if_exists="append", index=False)
    con.commit()
    total = con.execute("SELECT COUNT(*) FROM fx_rates").fetchone()[0]
    con.close()
    return total

raw = extract();      print(f"extract:   pulled {len(raw['rows'])} currencies for {raw['as_of']}")
df = validate(raw);   print(f"validate:  schema gate passed, {len(df)} rows are clean")
df = transform(df);   print("transform: sample ->", df.iloc[0].to_dict())
total = load(df);     print(f"load:      table now holds {total} rows total")

▶ Output (run 1, then run 2)

=== RUN 1 ===
extract:   pulled 4 currencies for 2026-07-09
validate:  schema gate passed, 4 rows are clean
transform: sample -> {'currency': 'EUR', 'rate': 0.87451, 'as_of': '2026-07-09', 'usd_per_unit': 1.1435}
load:      table now holds 4 rows total
=== RUN 2 ===
extract:   pulled 4 currencies for 2026-07-09
validate:  schema gate passed, 4 rows are clean
transform: sample -> {'currency': 'EUR', 'rate': 0.87451, 'as_of': '2026-07-09', 'usd_per_unit': 1.1435}
load:      table now holds 4 rows total

What happened here: the data flows in one direction, extract to validate to transform to load, and each step hands its output to the next exactly like the arrows in the DAG. The important detail is the two runs: after running the pipeline twice, the table still holds 4 rows, not 8. That is idempotency, and it is the difference between a pipeline you can safely rerun and one that quietly doubles your data every time someone clicks “retry”. Let us look at why that matters and the classic interview trap that goes with it.

Idempotency and the execution_date Trap

Idempotent is a big word for a simple promise: running the task once and running it five times leave the database in the same state. Think of a light switch labeled “off”. Flip it to off once or flip it to off ten times, the light is still off. Our load task got this right by deleting the rows for today’s date before inserting today’s rows, so a rerun overwrites instead of piling on. The lazy version, a bare INSERT, would add duplicate rows every retry, and Airflow retries automatically, so duplicates are not a maybe, they are a when.

Now the interview trap. In Airflow, the date a run is “for” is not the wall-clock time it actually runs. A DAG scheduled @daily for July 9 does not fire at the start of July 9, it fires at the end of the interval, in the early hours of July 10, once July 9’s data is complete. That logical date is passed into your tasks (as the logical_date, formerly called execution_date).

The rule to remember: always key your reads and writes off that logical date, never off datetime.now(). If you filter your source query by “today” using the wall clock, a backfill of last month will happily pull this month’s data and write it under the wrong date. Use the date Airflow hands you and every rerun, backfill included, lands in the right place.

Data-Quality Gates as First-Class Tasks

A quality gate is a task whose only job is to say yes or no. Think of it like the security check at an airport: nothing gets on the plane until it passes, and one flagged bag stops the whole line. In this Airflow tutorial’s pipeline the validate task is that gate, built with pandera, and because it is a real task in the graph, a failure there marks the run red and skips everything downstream. No bad rows reach the database, and you get an alert instead of a silent corruption you discover a week later. Here is what happens when a batch arrives with a nonsense negative rate.

📄 gate_fail.py: a failing gate blocks the load

import pandas as pd
import pandera.pandas as pa
from pandera.pandas import Column, DataFrameSchema, Check

schema = DataFrameSchema({
    "currency": Column(str, Check.str_length(3, 3)),
    "rate": Column(float, Check.greater_than(0)),
})

bad = pd.DataFrame([
    {"currency": "EUR", "rate": 0.8745},
    {"currency": "INR", "rate": -95.39},   # nonsense value slipped in
])

loaded = False
try:
    schema.validate(bad, lazy=True)   # lazy=True collects all failures
    loaded = True
except pa.errors.SchemaErrors as e:
    print("GATE BLOCKED the batch. Failure cases:")
    print(e.failure_cases[["column", "check", "failure_case"]].to_string(index=False))

print("did downstream load run?", loaded)

▶ Output

GATE BLOCKED the batch. Failure cases:
column           check  failure_case
  rate greater_than(0)        -95.39
did downstream load run? False

What happened here: pandera checked every row against the schema, found the -95.39 rate that fails the greater_than(0) rule, and raised. Because the gate raised, loaded stayed False, which is exactly how Airflow behaves: a raised exception fails the task, and downstream tasks are skipped, not run on garbage. In a real DAG that failure shows up as a red square in the grid and can page whoever is on call. This is why teams put schema checks in tools like pandera or Great Expectations as their own tasks instead of burying a few assert lines inside the transform. The gate is visible, testable, and it stops the line.

From a Drift Alert to a Retraining DAG

Everything above was a data pipeline, but the same DAG shape from this Airflow tutorial powers Machine Learning Operations (MLOps). In the model drift tutorial we watched a model’s accuracy quietly decay as the world changed under it, and we raised a drift alert when it crossed a threshold. That alert is the perfect trigger for a second DAG whose job is to retrain. The shape is familiar: pull the latest labeled data, validate it through a schema gate, train a fresh model, evaluate it against the current champion, and only promote the new model if it actually wins. If it loses, the old model stays live and you get notified, no silent downgrade.

📄 dags/retrain.py: a drift alert triggers this DAG (sketch)

@dag(schedule=None, start_date=pendulum.datetime(2026, 7, 1, tz="UTC"),
     catchup=False, tags=["mlops", "retrain"])
def retrain_on_drift():

    @task
    def pull_fresh_data() -> str: ...        # last N days of labeled rows

    @task
    def validate_data(path: str) -> str: ... # same pandera gate idea

    @task
    def train(path: str) -> str: ...         # returns new model run id

    @task.branch
    def promote_or_keep(new_run: str) -> str:
        # compare new model vs current champion on a holdout set
        return "promote" if new_is_better(new_run) else "keep_champion"

    @task
    def promote(): ...                       # move the "production" alias
    @task
    def keep_champion(): ...                  # log and alert, change nothing

    decision = promote_or_keep(train(validate_data(pull_fresh_data())))
    decision >> [promote(), keep_champion()]

retrain_on_drift()

What happened here: this DAG has schedule=None, meaning it does not run on a clock. Instead your drift monitor triggers it through the Airflow API when accuracy drops. The @task.branch decorator is the new piece: it picks which downstream path runs, so a better model flows to promote and a worse one flows to keep_champion. Promotion just moves an alias like production onto the new model version, which is the same registry pattern from the MLflow tutorial. Drift detection, orchestration, and the model registry are three tools that click together into one self-healing loop, and Airflow is the glue in the middle.

Airflow, Prefect, or Dagster?

The concept that outlives any tool is orchestration: express your work as a dependency graph, run it on a schedule or a trigger, retry the flaky parts, and keep a full history of every run. Learn that idea and the specific library becomes a detail you can swap. Airflow is the name in most postings at the time of writing, which is why this Airflow tutorial anchors on it, but two modern alternatives are worth knowing.

Prefect feels the most like plain Python. You decorate normal functions with @flow and @task and it figures out the graph from how you call them, with no separate DAG file to register. Teams pick it when they want to add orchestration to existing Python code with the least ceremony, and its dynamic, run-time task creation is more natural than Airflow’s historically static graphs.

Dagster flips the model to be asset-first. Instead of describing tasks, you describe the data assets you want to exist (a table, a model, a report) and how each is produced, and Dagster works out the order. It bakes in data-quality checks and lineage, so it appeals to teams who think in terms of “which datasets are fresh and correct” rather than “which scripts ran”. The graph you build in any of the three is the same graph. Pick the tool your team will actually maintain, and the skills carry across.

Common Mistakes

Best Practices

What’s Next?

You built a real pipeline in this Airflow tutorial: extract from an API, block bad data at a schema gate, transform, load idempotently, and notify, all wired as a DAG that retries on its own and remembers every run. You also saw how the same shape turns a drift alert into a retraining pipeline, which is how MLOps teams keep models fresh without a human babysitting them at 3 AM. That is genuinely more than most job postings ask of an entry-level candidate.

Want the full picture? Browse the complete Python + AI/ML tutorial series home to see how orchestration fits alongside the data, training, and deployment tutorials.

More in this series:

• LLM and GenAI Interview Questions: 40-Question Checkpoint
• AI Design Patterns: Routers, Fallbacks, Caches, and Agents
• From Idea to AI MVP: The Right Way to Build

Frequently Asked Questions

Interview Questions on Airflow

The ideas from this Airflow tutorial as they show up in real interviews, framed as scenarios you can practice out loud.

Q: What is a DAG and why does Airflow use one?

A DAG (Directed Acyclic Graph) is a set of tasks with directed arrows showing which must finish before which, and no cycles, so nothing ever depends on itself. Airflow uses it to know the correct order to run tasks, which ones can run in parallel, and where to stop when something upstream fails. It is the structure that lets Airflow retry a single failed step and skip everything downstream of it.

Q: Explain execution_date (logical_date). Why not use datetime.now()?

The logical_date is the date the run represents, not when it executes. A daily run for July 9 actually fires early on July 10, after July 9’s data is complete. If you filter source data by datetime.now(), a backfill of last month pulls today’s data instead of the historical day, so records land under the wrong date. Keying everything off the logical_date Airflow passes in makes reruns and backfills correct.

Q: What does it mean for a task to be idempotent, and how do you make a load idempotent?

Idempotent means running it once or many times leaves the same end state. Since Airflow retries and you backfill, non-idempotent tasks corrupt data. For a load, delete the rows for the run’s date and then insert, or use an upsert with a primary key, so a rerun replaces that day’s rows instead of appending duplicates.

Q: Your DAG runs fine with airflow dags test but the scheduler is slow to pick up changes. What do you check?

The scheduler parses every DAG file repeatedly, so expensive work at module level (top-level API calls, heavy imports, big computations outside task functions) slows parsing for every DAG. Move all of that inside the task functions, keep the file’s top level to lightweight definitions, and check the DAG parse time in the UI. Also confirm the file is actually in the dags folder and free of import errors.

Q: How would you wire a model-retraining pipeline that only runs when drift is detected?

Give the retraining DAG schedule=None so it does not run on a clock, and have the drift monitor trigger it through the Airflow API when accuracy crosses a threshold. Inside, pull fresh labeled data, validate it through a schema gate, train a new model, then use a branch task to compare it against the current champion on a holdout set. Promote the new model (move the production alias) only if it wins, otherwise keep the champion and alert. That keeps a bad model from ever going live automatically.

Go deeper: when you outgrow this post, Apache Airflow documentation is the next stop.

Related Posts

Previous: Python ML Production Deployment at Scale: FastAPI, Docker, Cloud

Next: AI Design Patterns: Routers, Fallbacks, Caches, and Agents

Series Home: Python + AI/ML Tutorial Series