The Complete Airflow Guide

This is part one of three. It covers everything you need to do real work with Airflow, not a teaser. By the end you can write a DAG, schedule it, pass data between tasks, retry failures sensibly, connect to a database, backfill history, and read the interface well enough to debug a red run. Mid-level and Senior take the same topics further; nothing here is thrown away.

Each section ends with a Try it task. Do them as you go. They take a few minutes each, and these concepts only stick once you have watched your own DAG turn green, then deliberately red.

What Airflow is, and the problem it solves

Airflow is a platform for writing, scheduling, and monitoring workflows in Python. You define a workflow as code, Airflow works out the order tasks must run in, executes them on a schedule, retries the ones that fail, and gives you a web interface showing what happened.

The diagram is the whole idea. Compare it to what came before.

Before this, a data pipeline was a chain of cron jobs. 0 2 * * * /opt/etl/extract.sh at 2am, then 0 3 * * * /opt/etl/transform.sh at 3am, and a hope that extract finished within the hour. When it did not, transform ran on yesterday's data and nobody noticed until a dashboard looked wrong on Thursday. There was no dependency graph, no retry, no history, and no answer to "did last Tuesday's run succeed?" other than grepping logs on whichever box happened to have them.

Airflow replaces the hope with a dependency graph. Transform does not run at 3am; it runs when extract has succeeded. If extract fails, transform does not run at all, and you get told.

Two consequences of the design explain most of what follows. Notice them now rather than discovering them later.

Workflows are Python files. That means your pipeline is versioned, reviewed, and diffed like any other code, and it also means Airflow executes your file to discover the DAG, which has consequences for performance and safety that we come back to repeatedly.

Airflow is an orchestrator, not a compute engine. It decides what runs and when, and it is deliberately bad at processing data itself. A task that loads a 40 GB dataframe into the worker's memory is a misuse; a task that tells Spark or BigQuery or dbt to do that work is correct. This distinction separates people who have operated Airflow from people who have only read about it.

What people use it for:

You need little to follow along: Python, Docker, and a terminal. The official Docker Compose setup gets you a working Airflow in about five minutes, and a DAG with three EmptyOperator tasks is enough to learn the whole scheduling model.

The architecture: five components, two planes

You will hit error messages that only make sense if you know which component is complaining, so learn the shape now.

Two things about this picture matter immediately.

The scheduler re-reads your DAG files continuously. By default every thirty seconds it scans the folder for changes, and it executes each file to find DAG objects. Slow top-level code therefore slows down the whole scheduler, which is the single most common cause of "why is Airflow so sluggish", and the reason for a rule we will state properly in a moment.

The metadata database is the source of truth, not your DAG file. The file says what should exist; the database records what did happen. When the UI shows a task as failed and your file no longer contains that task, both are correct. They are answering different questions.

There is a sixth component you will meet at Mid level, the triggerer, which handles deferred tasks. Ignore it for now.

Your first DAG, line by line

Enough theory. Create a file called hello_dag.py in your dags/ folder.

from datetime import datetime
from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="hello_world",
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["tutorial"],
) as dag:

    start = EmptyOperator(task_id="start")

    say_hello = BashOperator(
        task_id="say_hello",
        bash_command="echo 'Hello from Airflow'",
    )

    show_date = BashOperator(
        task_id="show_date",
        bash_command="date",
    )

    start >> say_hello >> show_date

Wait up to thirty seconds and it appears in the UI. Unpause it with the toggle and watch it run.

Now let me walk through every line, because this small file contains the entire conceptual model.

with DAG(...) as dag: creates the DAG object. Everything defined inside the block is attached to it automatically, which is why you rarely see dag=dag on each task in modern code.

dag_id="hello_world" is the unique identifier. It appears in the UI, in the CLI, and in the database, so renaming it creates a new DAG and abandons the old one's history.

start_date=datetime(2024, 1, 1) is the date from which Airflow considers intervals to exist, not the day you wrote the file. It is the single most misunderstood parameter in Airflow and it gets its own section shortly.

schedule="@daily" is how often it runs. A cron string, a preset like this, or None for manual-only.

catchup=False tells Airflow not to run every interval between start_date and now. Leave this out on a DAG with a 2022 start date and you will trigger seven hundred runs the moment you unpause it. More on this shortly too.

tags=["tutorial"] groups DAGs in the UI. Free, and worth using from the first DAG.

EmptyOperator does nothing. It exists as a marker, a clean entry or exit point in a graph, and it is useful.

BashOperator runs a shell command on the worker.

start >> say_hello >> show_date sets the dependencies. That >> is Airflow's bitshift operator, overloaded to mean "then".

Tasks, operators, and the dependency graph

Three words that get used loosely and mean different things.

So BashOperator is the operator, say_hello is the task, and "say_hello for 2024-05-01" is the task instance. The UI shows task instances; your file defines tasks.

The operators you will use early:

Dependencies have four shapes worth knowing:

a >> b                      # a, then b
a << b                      # b, then a  (the same thing, read backwards)

a >> [b, c] >> d            # fan out to b and c in parallel, then fan in to d

chain(a, b, c, d)           # a >> b >> c >> d, without the arrow soup
cross_downstream([a, b], [c, d])   # every one of a,b feeds every one of c,d

from airflow.models.baseoperator import chain, cross_downstream

The TaskFlow API: DAGs that read like Python

Everything above is the classic operator style, and you will read a lot of it. For Python work, modern Airflow has a cleaner form.

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

@dag(
    dag_id="taskflow_example",
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["tutorial"],
)
def etl_pipeline():

    @task
    def extract():
        return {"records": [1, 2, 3, 4, 5]}

    @task
    def transform(data: dict):
        return {"total": sum(data["records"])}

    @task
    def load(result: dict):
        print(f"Loading total: {result['total']}")

    load(transform(extract()))

etl_pipeline()

Read the last line inside the function: load(transform(extract())). That single expression does two things at once. It declares the dependency chain and passes the return values along it. No >>, no manual data passing.

They mix freely in one DAG, which is what you will do:

    raw = extract()                                    # TaskFlow task
    notify = BashOperator(task_id="notify", bash_command="echo done")
    raw >> notify                                      # classic dependency

Scheduling: the interval model, precisely

This is where almost everyone gets confused. Slow down: every later concept depends on it.

Airflow schedules intervals, not moments. A @daily DAG with start_date=2024-05-01 produces a run for the interval covering 1 May, and that run starts at the end of the interval, just after midnight on 2 May. The data for 1 May is only complete once 1 May is over.

The parameter values, in the order they cause trouble:

schedule="@daily"                    # midnight, every day
schedule="0 2 * * *"                 # 02:00 every day, cron
schedule="0 9 * * 1-5"               # 09:00 on weekdays
schedule=timedelta(hours=6)          # every six hours from start_date
schedule=None                        # manual and triggered runs only
schedule="@once"                     # exactly once

The other detail that costs people an afternoon: use the interval, not datetime.now(). A task that queries "yesterday" by calling now() produces different results on a rerun than it did originally, which destroys reproducibility. Airflow gives you the interval as a template variable.

BashOperator(
    task_id="extract",
    bash_command="python extract.py --date {{ ds }}",     # 2024-05-01
)

Retries, timeouts, and default_args

A pipeline that fails permanently on a transient network blip is not production-ready. Airflow's retry handling is a few parameters, and setting them is the difference between a pipeline you trust and one you babysit.

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator

default_args = {
    "owner": "data-platform",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(minutes=45),
    "email_on_failure": False,
}

with DAG(
    dag_id="robust_pipeline",
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False,
    default_args=default_args,
    dagrun_timeout=timedelta(hours=2),
    max_active_runs=1,
    tags=["production"],
) as dag:

    fetch = BashOperator(
        task_id="fetch_api",
        bash_command="python fetch.py --date {{ ds }}",
        retries=5,                          # overrides the default for this task
    )

default_args is applied to every task in the DAG, and any task can override an individual value. That pattern (sensible defaults at the DAG level, exceptions at the task level) is how real DAGs stay readable.

execution_timeout deserves special mention because its absence is the classic outage: a task that hangs on a network read holds a worker slot indefinitely, and with max_active_runs unset the next intervals pile up behind it until the whole DAG is stuck. One timedelta prevents it.

Passing data with XCom, and its limits

Tasks are separate processes, often on separate machines. XCom, cross-communication, is how they pass small values.

@task
def get_record_count():
    return 4_213                       # pushed to XCom automatically

@task
def report(count: int):
    print(f"Processed {count} records")

report(get_record_count())

With TaskFlow, returning a value pushes it and consuming it pulls it. With classic operators you do it by hand:

def _extract(**context):
    context["ti"].xcom_push(key="row_count", value=4213)

def _report(**context):
    count = context["ti"].xcom_pull(task_ids="extract", key="row_count")
    print(f"Processed {count} records")

In a template, ti.xcom_pull is available directly:

BashOperator(
    task_id="notify",
    bash_command="echo 'rows: {{ ti.xcom_pull(task_ids=\"extract\") }}'",
)

Connections and Variables: config outside the DAG

Credentials do not belong in your DAG file. Airflow has two stores for configuration, and they have different jobs.

Connections describe how to reach an external system. Create one in Admin → Connections, or from the CLI:

airflow connections add 'warehouse' \
    --conn-type 'postgres' \
    --conn-host 'db.internal' \
    --conn-port 5432 \
    --conn-schema 'analytics' \
    --conn-login 'airflow' \
    --conn-password 'secret'

from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

# In an operator: name the connection, never the credentials
SQLExecuteQueryOperator(
    task_id="daily_rollup",
    conn_id="warehouse",
    sql="sql/daily_rollup.sql",
)

# In Python: a hook gives you a real client
@task
def count_rows():
    hook = PostgresHook(postgres_conn_id="warehouse")
    return hook.get_first("SELECT count(*) FROM events")[0]

Variables hold configuration values that are not credentials, like a bucket name, a threshold, or a feature flag:

from airflow.models import Variable

bucket = Variable.get("data_bucket")
config = Variable.get("pipeline_config", deserialize_json=True)
retries = Variable.get("max_retries", default_var=3)

Templating with Jinja: values resolved at run time

That {{ ds }} is Jinja. Airflow renders templated fields at run time, which is how a task knows which interval it is working on.

BashOperator(
    task_id="load_partition",
    bash_command=(
        "python load.py "
        "--source s3://{{ var.value.data_bucket }}/raw/{{ ds }}/ "
        "--target events_{{ ds_nodash }} "
        "--run-id {{ dag_run.run_id }}"
    ),
)

SQLExecuteQueryOperator(
    task_id="rollup",
    conn_id="warehouse",
    sql="sql/rollup.sql",              # the FILE is templated too
    params={"lookback_days": 7},
)

DELETE FROM daily_stats WHERE day = '{{ ds }}';

INSERT INTO daily_stats (day, events, users)
SELECT '{{ ds }}', count(*), count(DISTINCT user_id)
FROM events
WHERE event_date BETWEEN '{{ macros.ds_add(ds, -params.lookback_days) }}'
                     AND '{{ ds }}';

Note the shape of that SQL: a DELETE for the target partition followed by an INSERT. That is the idempotent pattern. Rerun it as many times as you like and the result is identical, which is what makes retries and backfills safe.

Two things to know about templating:

Only declared fields are templated. Each operator has a template_fields list. BashOperator templates bash_command and env; SQLExecuteQueryOperator templates sql and parameters. Putting {{ ds }} in an untemplated field gives you the literal string.

macros gives you date arithmetic without importing anything: {{ macros.ds_add(ds, -7) }}, {{ macros.datetime.now() }}, {{ macros.ds_format(ds, "%Y-%m-%d", "%d/%m/%Y") }}.

Sensors: waiting without wasting a worker

Sometimes a task cannot start until something external exists: a file lands, a table is populated, an upstream team finishes. A sensor is an operator that waits.

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.sensors.filesystem import FileSensor

wait_for_file = S3KeySensor(
    task_id="wait_for_extract",
    bucket_name="raw-data",
    bucket_key="events/{{ ds }}/_SUCCESS",
    aws_conn_id="aws_default",
    poke_interval=60,                    # check every minute
    timeout=60 * 60 * 4,                 # give up after four hours
    mode="reschedule",                   # ← the important one
)

The parameter that matters most is mode, and getting it wrong is one of the classic ways to deadlock a cluster:

Two more waiting patterns worth knowing now:

from airflow.sensors.external_task import ExternalTaskSensor

# Wait for a task in a different DAG's matching interval
wait_upstream = ExternalTaskSensor(
    task_id="wait_for_ingest",
    external_dag_id="ingest_events",
    external_task_id="load_complete",
    mode="reschedule",
    timeout=60 * 60 * 2,
)

from airflow.sensors.python import PythonSensor

@task.sensor(poke_interval=120, timeout=3600, mode="reschedule")
def wait_for_rows():
    hook = PostgresHook(postgres_conn_id="warehouse")
    count = hook.get_first("SELECT count(*) FROM staging WHERE day = %s", ("{{ ds }}",))[0]
    return count > 0                      # truthy return means "done waiting"

Branching and trigger rules

Not every run should do the same thing. Airflow's branching operator picks a path at run time.

from airflow.operators.python import BranchPythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule

def _choose_path(**context):
    count = context["ti"].xcom_pull(task_ids="extract", key="return_value")
    return "full_reload" if count > 100_000 else "incremental_load"

branch = BranchPythonOperator(task_id="choose_path", python_callable=_choose_path)

full = EmptyOperator(task_id="full_reload")
incr = EmptyOperator(task_id="incremental_load")

join = EmptyOperator(
    task_id="join",
    trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,   # ← required
)

extract >> branch >> [full, incr] >> join

The branching callable returns a task_id, or a list of them, and Airflow skips every other branch. That skipping is why the join task needs a trigger rule: by default a task requires all upstreams to have succeeded, and a skipped upstream is not a success.

The TaskFlow equivalent is shorter:

@task.branch
def choose_path(count: int) -> str:
    return "full_reload" if count > 100_000 else "incremental_load"

all_done is the rule for cleanup tasks that must run regardless:

cleanup = BashOperator(
    task_id="cleanup",
    bash_command="rm -rf /tmp/staging/{{ ds }}",
    trigger_rule=TriggerRule.ALL_DONE,      # even if everything upstream failed
)

Reading the UI: Grid, Graph, and logs

The interface is your primary debugging tool, and knowing which view answers which question saves real time.

The colours you need to recognise:

The four actions on a task instance that you will use constantly:

The CLI and backfills: running history on purpose

The UI is convenient; the CLI is what you use when you need to be precise, and it is how you run a backfill.

# Inspect
airflow dags list
airflow dags list-runs -d my_dag
airflow tasks list my_dag --tree

# Validate before deploying
python dags/my_dag.py                       # does the file even import?
airflow dags list-import-errors

# Test a single task without touching the database
airflow tasks test my_dag my_task 2024-05-01

# Test a whole DAG run locally
airflow dags test my_dag 2024-05-01

# Trigger and manage
airflow dags trigger my_dag
airflow dags trigger my_dag --conf '{"reprocess": true}'
airflow dags pause my_dag
airflow dags unpause my_dag

# Backfill a date range
airflow dags backfill my_dag \
    --start-date 2024-04-01 \
    --end-date 2024-04-30

# Clear a range of task instances
airflow tasks clear my_dag \
    --start-date 2024-04-01 --end-date 2024-04-30 \
    --task-regex "transform.*" --downstream

Two commands deserve emphasis because they change how you work.

airflow tasks test runs one task instance immediately, in the foreground, printing logs to your terminal, and it does not record state in the metadata database. It is the fastest possible feedback loop for debugging a task, and it works on any date.

airflow dags backfill creates runs for a historical range. It is how you populate a new DAG's history, or reprocess after fixing a bug. It only works properly if your tasks are idempotent, which is why that property keeps coming up.

Top-level code, and the rule that matters most

This is the single most important performance rule in Airflow, and it follows directly from the architecture: the scheduler executes every DAG file, repeatedly, just to find the DAG objects.

# WRONG — this runs on every single parse
import pandas as pd
config = requests.get("https://config.internal/pipeline").json()
tables = PostgresHook("warehouse").get_records("SELECT name FROM tables")

with DAG(...) as dag:
    for table in tables:
        ...

# RIGHT — the expensive work happens at run time
with DAG(...) as dag:

    @task
    def get_tables():
        import pandas as pd                        # imported inside the task
        return PostgresHook("warehouse").get_records("SELECT name FROM tables")

    @task
    def process(table: str):
        ...

    process.expand(table=get_tables())             # dynamic mapping, Mid level

The cost is easy to underestimate. A Variable.get() at the top level of one DAG is one query every thirty seconds, 2,880 a day. Across twenty DAG files that is 57,600 queries a day for values nobody read. An API call up there adds its latency to every parse cycle.

# Measure it
airflow dags list-import-errors
airflow dags report                     # parse duration per file

The Admin → DAG Processing page shows the same thing in the UI: file name, number of DAGs, and parse duration. Anything over a second deserves a look; anything over five needs fixing.

Putting it all together

Everything above in one DAG. Nothing here is new. Read it as a whole and you should be able to justify every line.

"""Daily events pipeline: land → validate → transform → publish → notify."""
from datetime import datetime, timedelta

from airflow.decorators import dag, task
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.sensors.filesystem import FileSensor
from airflow.utils.trigger_rule import TriggerRule

# Sensible defaults for every task; individual tasks override as needed.
default_args = {
    "owner": "data-platform",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "execution_timeout": timedelta(minutes=30),
}

@dag(
    dag_id="daily_events_pipeline",
    start_date=datetime(2024, 1, 1),
    schedule="0 2 * * *",              # 02:00, after the source system settles
    catchup=False,                     # backfill deliberately, never accidentally
    default_args=default_args,
    dagrun_timeout=timedelta(hours=3), # the whole run has a ceiling
    max_active_runs=1,                 # no overlapping runs on the same table
    tags=["production", "events"],
    doc_md=__doc__,                    # this docstring renders in the UI
)
def daily_events_pipeline():

    start = EmptyOperator(task_id="start")

    # reschedule mode: releases the worker slot between checks.
    # timeout: never wait forever.
    wait_for_extract = FileSensor(
        task_id="wait_for_extract",
        filepath="/data/landing/events_{{ ds_nodash }}.parquet",
        poke_interval=120,
        timeout=60 * 60 * 2,
        mode="reschedule",
    )

    @task
    def validate() -> int:
        """Row count is returned through XCom — a number, not the data."""
        hook = PostgresHook(postgres_conn_id="warehouse")   # a connection, not a password
        count = hook.get_first(
            "SELECT count(*) FROM staging.events WHERE day = %s",
            parameters=("{{ ds }}",),
        )[0]
        if count == 0:
            raise ValueError("No rows landed for {{ ds }}")
        return count

    # Idempotent by construction: DELETE the partition, then INSERT it.
    transform = SQLExecuteQueryOperator(
        task_id="transform",
        conn_id="warehouse",
        sql="sql/daily_rollup.sql",      # templated file, kept out of the DAG
        params={"lookback_days": 7},
    )

    @task
    def publish(row_count: int) -> str:
        """Returns a path, not a payload."""
        target = f"s3://analytics/daily/{{{{ ds }}}}/events.parquet"
        print(f"Published {row_count} rows to {target}")
        return target

    notify = BashOperator(
        task_id="notify",
        bash_command='echo "events {{ ds }} done: {{ ti.xcom_pull(task_ids=\'publish\') }}"',
    )

    # ALL_DONE: cleanup must happen whether the run succeeded or not.
    cleanup = BashOperator(
        task_id="cleanup",
        bash_command="rm -f /tmp/events_{{ ds_nodash }}.tmp",
        trigger_rule=TriggerRule.ALL_DONE,
    )

    end = EmptyOperator(task_id="end")

    row_count = validate()
    start >> wait_for_extract >> row_count >> transform
    transform >> publish(row_count) >> notify >> cleanup >> end

daily_events_pipeline()

-- Idempotent: safe to retry, safe to backfill.
DELETE FROM analytics.daily_events WHERE day = '{{ ds }}';

INSERT INTO analytics.daily_events (day, events, users, revenue)
SELECT
    '{{ ds }}'                        AS day,
    count(*)                          AS events,
    count(DISTINCT user_id)           AS users,
    sum(amount)                       AS revenue
FROM staging.events
WHERE event_date BETWEEN '{{ macros.ds_add(ds, -params.lookback_days) }}'
                     AND '{{ ds }}';

Twelve decisions in there are the whole lesson of this page:

What you can now do, and what comes next

You can write a DAG in both styles, reason correctly about intervals and logical dates, set retries and timeouts that make a pipeline trustworthy, pass metadata between tasks without abusing the database, keep credentials out of your code, template dates and SQL, wait on external systems without deadlocking your workers, branch conditionally and handle the trigger rules that come with it, read every view in the UI, use the CLI for fast feedback and deliberate backfills, and keep the scheduler fast by understanding top-level code. That is a working practitioner's toolkit, enough to own real DAGs on a shared Airflow.

Mid-level takes every one of those topics further: the executors and what each one costs, dynamic task mapping with .expand(), deferrable operators that replace sensors entirely, TaskGroups and DAG factories, datasets and data-aware scheduling, pools and priority weights for concurrency control, SLAs and callbacks, custom operators and hooks, testing DAGs properly, and the CI patterns that stop a broken DAG reaching production.

Senior then covers what you own when Airflow is your responsibility: the security model and who can see which DAG, secrets backends so credentials never live in the metadata database, multi-tenancy and isolation between teams, scaling the scheduler and the database, cost control across executors, upgrade and migration strategy, observability and SLOs for a platform, incident playbooks for a stuck scheduler and a poisoned queue, and where Airflow stops and a streaming or in-warehouse tool begins.

Part one of three. A fast review of everything in the Beginner Detailed track, in about twenty-five minutes. Fast review first, common questions at the end. Mid-level reviews this plus its own material; Senior reviews all three.

The thirty-second answer

Airflow is a platform for writing, scheduling, and monitoring workflows as Python code. You define a DAG, a directed acyclic graph of tasks, and Airflow works out the execution order, runs it on a schedule, retries failures, and records every run in a metadata database with a UI on top. It replaces a chain of cron jobs that had no dependency graph, no retries, and no history.

Then add the sentence that shows you have used it: "the thing people get wrong is that Airflow is an orchestrator, not a compute engine: a task should tell Spark or the warehouse to do the work, not load a dataframe into the worker."

The architecture

Two facts to state unprompted: the scheduler executes every DAG file repeatedly, so slow top-level code slows the whole scheduler, and the metadata database is the source of truth, not your file. The file says what should exist; the database records what happened.

Vocabulary

Your first DAG, annotated

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator

with DAG(
    dag_id="hello_world",              # unique; renaming abandons its history
    start_date=datetime(2024, 1, 1),   # the first interval considered
    schedule="@daily",                 # cron, preset, timedelta, or None
    catchup=False,                     # do NOT run every missed interval
    default_args={"retries": 3, "retry_delay": timedelta(minutes=5)},
    tags=["tutorial"],
) as dag:
    start = EmptyOperator(task_id="start")
    hello = BashOperator(task_id="hello", bash_command="echo hi {{ ds }}")
    start >> hello                     # ">>" means "then"

Mandatory: dag_id, start_date, and per task a task_id. Everything else has a default, and two of those defaults will hurt you, which is why catchup=False and explicit retries appear in every example.

TaskFlow versus classic operators

@dag(dag_id="etl", start_date=datetime(2024, 1, 1), schedule="@daily", catchup=False)
def etl():
    @task
    def extract(): return {"rows": [1, 2, 3]}

    @task
    def transform(d: dict): return {"total": sum(d["rows"])}

    @task
    def load(r: dict): print(r["total"])

    load(transform(extract()))         # dependencies AND data passing, one line
etl()

They mix freely. Use TaskFlow for Python you wrote; use a provider operator for "run this SQL" or "submit this Spark job".

Scheduling: the interval model

Airflow schedules intervals, not moments. A @daily run for 1 May starts just after midnight on 2 May, because the data for 1 May is only complete once 1 May is over.

Retries, timeouts, and idempotence

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "execution_timeout": timedelta(minutes=45),   # kill a hung task
}

@dag(..., default_args=default_args,
     dagrun_timeout=timedelta(hours=2),          # cap the whole run
     max_active_runs=1)                          # no overlapping writers

-- Idempotent by construction: safe to retry, safe to backfill.
DELETE FROM daily_stats WHERE day = '{{ ds }}';
INSERT INTO daily_stats SELECT '{{ ds }}', count(*) FROM events
WHERE event_date = '{{ ds }}';

XCom

@task
def count() -> int: return 4213        # pushed automatically

@task
def report(n: int): print(n)           # pulled automatically

report(count())

XCom is serialised into the metadata database. That is the whole reason for the rule: bloating it slows every UI page and the scheduler loop. Write the data to storage and pass the path.

Connections and Variables

airflow connections add warehouse --conn-type postgres \
  --conn-host db.internal --conn-login airflow --conn-password secret

SQLExecuteQueryOperator(task_id="rollup", conn_id="warehouse", sql="sql/rollup.sql")

@task
def count():
    return PostgresHook(postgres_conn_id="warehouse").get_first("SELECT count(*) FROM events")[0]

Sensors

S3KeySensor(
    task_id="wait",
    bucket_key="events/{{ ds }}/_SUCCESS",
    poke_interval=60,
    timeout=60 * 60 * 4,        # ALWAYS set a timeout
    mode="reschedule",          # ← releases the worker slot between checks
)

Branching and trigger rules

@task.branch
def choose(count: int) -> str:
    return "full_reload" if count > 100_000 else "incremental_load"

join = EmptyOperator(
    task_id="join",
    trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,   # ← required
)

Reading the UI

"Clear" is the verb, not "rerun". You clear a task instance's state and the scheduler notices there is now a stateless task instance to run. That indirection explains why clearing works on old runs and can cascade downstream.

The CLI

airflow tasks test my_dag my_task 2024-05-01   # run one task NOW, no DB state
airflow dags test my_dag 2024-05-01            # a whole run, locally
python dags/my_dag.py                          # does it even import?
airflow dags list-import-errors
airflow dags backfill my_dag --start-date 2024-04-01 --end-date 2024-04-30
airflow tasks clear my_dag -s 2024-04-01 -e 2024-04-30 --downstream

airflow tasks test is the fastest feedback loop in Airflow: foreground execution, logs in your terminal, and no state recorded in the metadata database.

Top-level code: the rule that matters most

The one-line test: ask of every line outside a task: "do I want this to run every thirty seconds forever?" If no, it belongs inside a task.

Common interview questions

1. What is Airflow and what problem does it solve?A platform for writing, scheduling, and monitoring workflows as Python code. It gives you a dependency graph, retries, a scheduling model with historical intervals, and a UI with full run history, replacing cron chains that had none of those and no answer to "did last Tuesday work?".
2. What is a DAG?A directed acyclic graph of tasks. Directed because dependencies point one way; acyclic because a task cannot eventually depend on itself, which Airflow enforces as an import error. That guarantee is what lets the scheduler always compute an execution order.
3. Name the components and what each does.Scheduler parses DAG files and decides what is ready. Executor hands queued tasks to workers. Workers run the code. The metadata database holds all state. The webserver serves the UI and API. The triggerer holds deferred tasks. The webserver going down does not stop pipelines running.
4. Operator versus task versus task instance?The operator is the class, a kind of work. The task is an instance of it in a DAG with a task_id. The task instance is one run of that task for one specific interval. The UI shows task instances; your file defines tasks.
5. Why does a @daily DAG for 1 May run on 2 May?Airflow schedules intervals, not moments, and a run fires at the end of its interval, because the data for 1 May is only complete once 1 May is over. The logical date is 2024-05-01 even though the wall clock says the second.
6. What does catchup=True do, and why is it dangerous?It creates a run for every interval between start_date and now. With a two-year-old start date and a daily schedule that is around eight hundred runs the moment you unpause, which can saturate your cluster and hammer every downstream system. Default to False and backfill deliberately.
7. Why use {{ ds }} instead of datetime.now()?Because a rerun or a backfill must produce the same result as the original run. now() makes the output depend on when you ran it, so clearing an old task instance silently reprocesses the wrong window.
8. What is XCom, and what should not go in it?The mechanism for passing small values between tasks. It is serialised into the metadata database, so row counts, paths, and versions are fine while dataframes and file contents are not. Those bloat the database and slow every UI page. Write the data to storage and pass the path.
9. Where do credentials belong?In a Connection, referenced by conn_id from an operator or a Hook. Never in the DAG file, and never in a Variable you print. Non-secret configuration goes in a Variable.
10. Why must you not call Variable.get() at the top level?Top-level code executes on every parse cycle (every thirty seconds, for every DAG file) so it becomes thousands of database queries a day for a value nobody read. Worse, an API call up there adds its latency to every parse and can stall the scheduler.
11. What makes a task idempotent, and why does it matter?Running it twice produces the same result as once. It matters because Airflow retries, and because you will eventually backfill. The standard pattern is deleting the target partition then writing it, rather than appending.
12. What is a sensor, and what is the mode trap?An operator that waits for an external condition. In the default poke mode it holds a worker slot for the entire wait, so a handful of long waits can deadlock a cluster. mode="reschedule" releases the slot between checks, and always set a timeout.
13. How do you make one branch of a DAG run conditionally?A branching operator whose callable returns the task_id to follow; every other branch is skipped. The catch is that any join afterwards needs an explicit trigger rule, usually none_failed_min_one_success. Otherwise the skip propagates and silently skips the rest of the DAG.
14. What trigger rule would you use for a cleanup task?all_done, so it runs whether the upstream work succeeded, failed, or was skipped. That is the case the default all_success cannot express.
15. How do you rerun a failed task?Clear its state, and the scheduler picks it up because there is now a task instance with no state. There is no rerun button. Understanding that indirection explains why clearing works on historical runs and why Clear + Downstream cascades.
16. What is the difference between execution_timeout and dagrun_timeout?The first kills an individual task that runs too long; the second fails the whole run. You want both: the task timeout stops a hang holding a worker slot, and the run timeout stops intervals piling up behind a stuck run.
17. How do you test a single task quickly?airflow tasks test dag_id task_id 2024-05-01. It runs in the foreground, streams logs to your terminal, and records no state in the metadata database, which is the fastest feedback loop available.
18. How do you backfill, and what is the risk?airflow dags backfill with a date range. The risk is that thirty days of a twenty-task DAG is six hundred task instances competing with live pipelines. Set max_active_runs, and only backfill something idempotent.
19. Why is Airflow not a good place to process data?It is an orchestrator. A task that loads a large dataframe into the worker consumes a scheduling slot for the duration, cannot scale beyond one machine, and puts memory pressure on a shared component. The task should tell Spark, the warehouse, or a container to do the work.
20. Your DAG is not showing up in the UI. Debug it.Is the file in the DAG folder with a .py extension; does it import at all, which python dags/my_dag.py answers; is there an entry in airflow dags list-import-errors; does the file create a DAG object at module level; and has the scheduler had thirty seconds to notice.

Sixty-second self-test

• Give the thirty-second answer, then the "orchestrator, not compute engine" sentence.
• Name the six components and one symptom each.
• Say what the scheduler does to your DAG file, and how often.
• Explain why a @daily run for 1 May starts on 2 May.
• Say what catchup=True with an old start_date does.
• Give the reason to use {{ ds }} over now().
• Say what XCom is for and what it must not carry.
• Name where credentials belong, and the top-level-code rule.
• Give the sensor mode that frees a worker slot, and the missing-timeout risk.
• Say why a join after a branch needs a trigger rule.
• Explain what "clear" does and why there is no rerun button.
• Define an idempotent task and give the SQL pattern for one.

Part one of three. Almost every beginner problem with Airflow comes from one of three things: misunderstanding the interval model, putting expensive work at the top level of a DAG file, or forgetting that a task must be safe to run twice. Start with the error table, then work through the habits and practice cards underneath it.

Common errors at this level

The habits that pay off most

Practice cards

Short, self-contained exercises. Each takes a few minutes and leaves you with a fact you will not forget.

1. Watch the interval modelCreate a @daily DAG with start_date three days ago and catchup=True. Count the runs, then read a log line printing both {{ ds }} and $(date -I). Seeing them differ is the moment scheduling clicks.
2. Flood your own clusterSet a two-year-old start_date with catchup=True on a scratch DAG and unpause it. Count the runs, then pause it immediately. Doing this once, deliberately, means never doing it accidentally.
3. Break idempotenceWrite a task that INSERTs rows. Clear it and rerun. Watch the row count double. Then rewrite it as delete-then-insert and repeat.
4. Deadlock a worker poolCreate more poke-mode sensors than you have worker slots, all waiting on files that do not exist. Watch real tasks starve. Then switch to reschedule.
5. Cause the silent skipBuild a branch with a join that has the default trigger rule. Watch the join and everything after it get skipped with no error anywhere.
6. Measure top-level costNote your parse duration in Admin → DAG Processing. Add time.sleep(3) at module level. Watch the number move. Move it inside a task and watch it recover.
7. Abuse XComReturn list(range(500_000)) from a task and observe the run duration and the XCom page. Then pass a file path instead and compare.
8. Kill a hung taskAdd a task running sleep 300 with execution_timeout=timedelta(seconds=20). Watch it be killed and read the message.
9. Tour every stateBuild one DAG that produces `success`, `failed`, `upstream_failed`, `skipped`, `up_for_retry`, and `up_for_reschedule`. Recognising all six on sight is worth the ten minutes.

Debugging order

Follow this rather than guessing. The first three steps answer most problems.

1. Does the file import?python dags/my_dag.py. Then airflow dags list-import-errors. A DAG that does not appear is almost always this.
2. Read the task logClick the failed task, then Logs. Scroll to the top of the traceback rather than the bottom. The last line is usually a wrapper, not the cause.
3. Check the rendered templateThe Rendered Template tab shows what your field became. Half of "my SQL is wrong" is a template that resolved differently than expected.
4. Run the task by handairflow tasks test my_dag my_task 2024-05-01. Foreground, instant, no state written. If it fails here too, you are debugging your code and not Airflow.
5. Check the state, not just the colourqueued means a capacity problem, upstream_failed means look upstream, skipped means a branch or trigger rule. Each points somewhere different.
6. Check the DAG's own healthAdmin → DAG Processing for parse time, and the Code view to confirm the scheduler is running the file you think it is.

python dags/my_dag.py                          # import check
airflow dags list-import-errors
airflow tasks test my_dag my_task 2024-05-01   # instant, no DB state
airflow dags test my_dag 2024-05-01            # a whole run, locally
airflow tasks list my_dag --tree               # the structure Airflow sees
airflow dags report                            # parse duration per file

The interval model, in practice

This is where most beginner time is lost, so start with these three questions.

"Why has today's run not happened?" Because the interval has not ended. A @daily DAG produces the run for 1 May just after midnight on 2 May.

"Which date should my task use?" {{ ds }}, the logical date, which is the start of the interval. Never the wall clock.

"How do I run yesterday again?" Clear that task instance. It will rerun with the original {{ ds }}, which is why idempotence matters.

Idempotence, concretely

"Safe to run twice" is easy to say and easy to get wrong. Three patterns cover almost everything.

DELETE FROM daily_stats WHERE day = '{{ ds }}';
INSERT INTO daily_stats SELECT '{{ ds }}', count(*) FROM events
WHERE event_date = '{{ ds }}';

MERGE INTO daily_stats t
USING (SELECT '{{ ds }}' AS day, count(*) AS n FROM events
       WHERE event_date = '{{ ds }}') s
ON t.day = s.day
WHEN MATCHED THEN UPDATE SET events = s.n
WHEN NOT MATCHED THEN INSERT (day, events) VALUES (s.day, s.n);

@task
def write_partition():
    path = "s3://analytics/events/dt={{ ds }}/"
    # write mode "overwrite" on the partition, not "append"
    df.write.mode("overwrite").parquet(path)

For non-idempotent side effects (sending a notification, charging a card) the pattern is to make the decision idempotent: record in a table that the notification for {{ ds }} was sent, and check that before sending.

Structuring a DAG that ages well

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

DEFAULTS = {
    "owner": "data-platform",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "execution_timeout": timedelta(minutes=30),
}

@dag(
    dag_id="pipeline",
    start_date=datetime(2024, 1, 1),
    schedule="0 2 * * *",
    catchup=False,
    default_args=DEFAULTS,
    dagrun_timeout=timedelta(hours=2),
    max_active_runs=1,
    tags=["production", "events"],
    doc_md="""
    ### Events pipeline
    **Owner:** data-platform · **Consumers:** exec dashboard
    **On failure:** idempotent — safe to clear and rerun.
    **Escalate if:** not green by 07:00 UTC.
    """,
)
def pipeline():
    ...
pipeline()

Local development

# Get an Airflow you can break
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'
mkdir -p dags logs plugins config
echo -e "AIRFLOW_UID=$(id -u)" > .env
docker compose up airflow-init && docker compose up -d

# The fast loop — no UI needed
airflow tasks test my_dag my_task 2024-05-01
airflow dags test my_dag 2024-05-01

# Before you commit
python dags/my_dag.py
airflow dags list-import-errors

Small things worth doing from day one

Set tags on every DAG. The UI filter is the only thing that makes fifty DAGs navigable.

Put the owner in default_args. It shows in the UI and answers "who do I ask?" without a search.

Write doc_md. Even three lines. It renders in the UI where the on-call person is already looking.

Prefer EmptyOperator bookends. A start and end task make the Graph view readable and give you a stable place to attach new work.

Never rename a dag_id casually. It is the identity: renaming abandons all history and creates a new DAG with none.

Grep for now() before every review. Each occurrence is a reproducibility bug or a deliberate exception, and there is no third option.

Check dvc-style: does a second run change anything? Run your pipeline twice on the same interval and diff the output. If it differs, the task is not idempotent.

# A starter default_args worth copying
DEFAULTS = {
    "owner": "your-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(minutes=30),
}

A starter DAG worth keeping

Every line here is something from this page.

"""
### Starter pipeline
**Owner:** your-team · **Consumers:** nobody yet
**On failure:** idempotent — safe to clear and rerun any interval.
"""
from datetime import datetime, timedelta

from airflow.decorators import dag, task
from airflow.operators.empty import EmptyOperator
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.sensors.filesystem import FileSensor
from airflow.utils.trigger_rule import TriggerRule

DEFAULTS = {
    "owner": "your-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "execution_timeout": timedelta(minutes=30),   # no task hangs forever
}

@dag(
    dag_id="starter",
    start_date=datetime(2024, 1, 1),
    schedule="0 2 * * *",
    catchup=False,                     # backfill deliberately, never by accident
    default_args=DEFAULTS,
    dagrun_timeout=timedelta(hours=2), # the whole run has a ceiling
    max_active_runs=1,                 # one writer per table
    tags=["starter", "dev"],
    doc_md=__doc__,                    # the runbook, in the UI
)
def starter():

    start = EmptyOperator(task_id="start")

    wait = FileSensor(
        task_id="wait_for_input",
        filepath="/data/landing/events_{{ ds_nodash }}.csv",
        poke_interval=60,
        timeout=60 * 60 * 2,           # never wait forever
        mode="reschedule",             # release the worker slot while waiting
    )

    @task
    def validate() -> int:
        """Logic lives in a module; the task just calls it."""
        from mypkg.checks import count_rows          # imported inside the task
        count = count_rows(day="{{ ds }}")
        if count == 0:
            raise ValueError("no rows for {{ ds }}")
        return count                                 # a number through XCom

    # Idempotent: DELETE the partition, then INSERT. SQL in a file.
    transform = SQLExecuteQueryOperator(
        task_id="transform",
        conn_id="warehouse",           # a Connection, never a password
        sql="sql/daily_rollup.sql",
    )

    cleanup = EmptyOperator(
        task_id="cleanup",
        trigger_rule=TriggerRule.ALL_DONE,   # runs whatever happened upstream
    )
    end = EmptyOperator(task_id="end")

    start >> wait >> validate() >> transform >> cleanup >> end

starter()

-- Safe to retry, safe to backfill.
DELETE FROM analytics.daily_events WHERE day = '{{ ds }}';
INSERT INTO analytics.daily_events (day, events)
SELECT '{{ ds }}', count(*) FROM staging.events WHERE event_date = '{{ ds }}';

Nine details in there are the whole lesson of this page: catchup=False, {{ ds }} rather than now(), idempotent delete-then-insert SQL in a file, retries with backoff, execution_timeout per task and dagrun_timeout per run, max_active_runs=1, a sensor in reschedule mode with a timeout, ALL_DONE on cleanup, and a heavy import moved inside the task.

Mid-level tips go deeper on every one of these: executors and the concurrency limit that binds, dynamic mapping pitfalls, deferrable operators replacing sensors, pools for shared systems, alerting people read, testing DAGs in CI, and diagnosing "it works locally but not on the cluster".

This is part two of three. It picks up exactly where Beginner ended and takes every topic from there further, then adds the machinery you have not met yet. Nothing is dropped and nothing is repeated for its own sake. Where you already know the basics, we go straight to the depth.

Where this picks up

Each section starts with the problem it solves, and ends with a Try it you can do on a real Airflow in a few minutes.

Executors, and where each one hurts

Beginner treated the executor as a black box. Choosing it is the single most consequential architectural decision in an Airflow deployment.

The trade-off that matters is startup latency versus isolation:

[core]
executor = CeleryExecutor
parallelism = 64                 # cluster-wide simultaneous task limit
max_active_tasks_per_dag = 16
max_active_runs_per_dag = 3
dag_file_processor_timeout = 60

[celery]
worker_concurrency = 16          # tasks per worker process

The three limits interact and people get caught by the tightest one:

The triggerer is the sixth component, and it exists for deferrable operators. It runs an asyncio event loop that can hold tens of thousands of waiting tasks in one process, which is the point of the next section but one.

Dynamic task mapping

Beginner's DAGs had a fixed task list. Real pipelines often need "one task per file that landed", and the count is unknown until run time.

@task
def list_files() -> list[str]:
    hook = S3Hook(aws_conn_id="aws_default")
    return hook.list_keys(bucket_name="raw", prefix="events/{{ ds }}/")

@task
def process(key: str) -> int:
    ...
    return rows

@task
def summarise(counts: list[int]):
    print(f"Total rows: {sum(counts)}")

counts = process.expand(key=list_files())     # N parallel task instances
summarise(counts)

.expand() creates one mapped task instance per element, at run time, and the UI groups them under a single task with an index. summarise receives the list of all results. That collection is automatic.

Three forms cover almost everything:

# One argument varies
process.expand(key=list_files())

# Several arguments vary together, zipped
process.expand(key=keys, region=regions)          # cross product!
process.expand_kwargs([{"key": "a", "region": "eu"}, {"key": "b", "region": "us"}])

# Some arguments fixed, some varying
process.partial(bucket="raw", retries=5).expand(key=list_files())

Mapping classic operators works too:

BashOperator.partial(task_id="process").expand(
    bash_command=["echo one", "echo two", "echo three"],
)

The limits are real, so know them:

[core]
max_map_length = 1024            # the cap on mapped instances per task

Deferrable operators and the triggerer

Beginner used sensors and learned that poke mode wastes worker slots. Deferrable operators remove the problem entirely.

A deferrable operator starts, discovers it must wait, and defers: it releases its worker slot, handing a lightweight Trigger to the triggerer process. The triggerer runs thousands of these in one asyncio loop. When the condition is met, the task is rescheduled onto a worker to finish.

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.sensors.time_delta import TimeDeltaSensorAsync

wait = S3KeySensor(
    task_id="wait_for_file",
    bucket_name="raw-data",
    bucket_key="events/{{ ds }}/_SUCCESS",
    deferrable=True,                     # ← the whole change
    poke_interval=60,
    timeout=60 * 60 * 4,
)

Many operators are deferrable, not just sensors. deferrable=True on a Spark submit, a BigQuery job, or an EMR step means Airflow does not hold a worker while an external system works.

Writing your own is a class and a trigger:

from airflow.triggers.base import BaseTrigger, TriggerEvent

class TableReadyTrigger(BaseTrigger):
    def __init__(self, table: str, day: str):
        super().__init__()
        self.table, self.day = table, day

    def serialize(self):
        return ("mypkg.triggers.TableReadyTrigger",
                {"table": self.table, "day": self.day})

    async def run(self):
        while True:
            if await self._rows_exist():
                yield TriggerEvent({"status": "ready"})
                return
            await asyncio.sleep(60)

class WaitForTable(BaseOperator):
    def execute(self, context):
        self.defer(
            trigger=TableReadyTrigger(self.table, context["ds"]),
            method_name="execute_complete",
        )

    def execute_complete(self, context, event=None):
        return event["status"]

Datasets and data-aware scheduling

Beginner scheduled on time. Sometimes the correct trigger is not "it is 2am" but "the upstream table was updated", and chaining DAGs with ExternalTaskSensor is fragile because it couples two schedules.

from airflow.datasets import Dataset

events_table = Dataset("postgres://warehouse/analytics.daily_events")
features = Dataset("s3://ml/features/latest.parquet")

# Producer: declares that a task updates a dataset
@dag(dag_id="build_events", schedule="0 2 * * *", start_date=..., catchup=False)
def build_events():
    @task(outlets=[events_table])
    def load():
        ...
    load()

# Consumer: scheduled by the dataset, not by the clock
@dag(dag_id="build_features", schedule=[events_table], start_date=..., catchup=False)
def build_features():
    @task(outlets=[features])
    def featurize():
        ...
    featurize()

# Waits for BOTH upstream datasets
@dag(dag_id="train_model", schedule=[events_table, features], ...)
def train_model():
    ...

The consumer runs when every dataset in its schedule list has been updated since its last run. No sensor, no polling, no shared schedule assumption, and the Datasets view in the UI draws the resulting cross-DAG graph.

You can also combine time and data conditions, and express logic across them:

from airflow.timetables.datasets import DatasetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable

@dag(
    schedule=DatasetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 6 * * *", timezone="UTC"),
        datasets=(events_table & features),        # AND; use | for OR
    ),
    ...
)

TaskGroups and DAG factories

A DAG with sixty tasks is unreadable in the Graph view. TaskGroups collapse related tasks into one expandable node.

from airflow.utils.task_group import TaskGroup

with DAG(...) as dag:
    start = EmptyOperator(task_id="start")

    with TaskGroup(group_id="extract") as extract:
        for source in ["orders", "users", "events"]:
            BashOperator(task_id=source, bash_command=f"python extract.py --src {source}")

    with TaskGroup(group_id="validate") as validate:
        schema = EmptyOperator(task_id="schema")
        rows = EmptyOperator(task_id="row_counts")
        schema >> rows

    start >> extract >> validate

TaskGroups are organisational. They create no runtime overhead. The task_id becomes extract.orders, which matters when you target a task from the CLI. The decorator form is neater in TaskFlow:

from airflow.decorators import task_group

@task_group(group_id="quality_checks")
def quality_checks(table: str):
    @task
    def check_nulls(t: str): ...
    @task
    def check_freshness(t: str): ...
    check_nulls(table) >> check_freshness(table)

A DAG factory generates similar DAGs from configuration, a common need when twenty tables follow the same pattern:

import yaml
from pathlib import Path

CONFIG = Path(__file__).parent / "config" / "tables.yaml"

def build_dag(name: str, cfg: dict):
    @dag(
        dag_id=f"load_{name}",
        start_date=datetime(2024, 1, 1),
        schedule=cfg["schedule"],
        catchup=False,
        tags=["generated", cfg["domain"]],
    )
    def _dag():
        SQLExecuteQueryOperator(
            task_id="load",
            conn_id=cfg["conn_id"],
            sql=f"sql/{name}.sql",
        )
    return _dag()

# Reading a local YAML at parse time is cheap and acceptable.
for name, cfg in yaml.safe_load(CONFIG.read_text()).items():
    globals()[f"load_{name}"] = build_dag(name, cfg)

Pools, priority, and concurrency control

Beginner set max_active_runs. That protects one DAG. Pools protect a shared resource across every DAG.

The classic problem: a legacy database accepts five concurrent connections. Ten DAGs each with a task hitting it will open forty and take it down.

airflow pools set legacy_db 5 "Legacy Oracle: max 5 concurrent connections"
airflow pools set gpu 2 "GPU nodes"
airflow pools list

SQLExecuteQueryOperator(
    task_id="legacy_extract",
    conn_id="legacy_oracle",
    sql="sql/extract.sql",
    pool="legacy_db",
    pool_slots=1,              # a heavy task can take several slots
    priority_weight=10,        # higher wins when the pool is contended
)

priority_weight has a detail worth knowing: by default it is downstream, meaning a task's effective weight includes its downstream tasks, so a task blocking a long chain naturally outranks a leaf. You can set weight_rule="absolute" to use the number literally.

SLAs, callbacks, and alerting people read

Beginner's DAG failed silently unless someone looked. Alerting is the difference between a pipeline you own and one that owns you.

def alert_failure(context):
    ti = context["task_instance"]
    send_slack(
        channel="#data-alerts",
        text=(
            f":red_circle: *{ti.dag_id}.{ti.task_id}* failed\n"
            f"Run: {context['run_id']}  Logical date: {context['ds']}\n"
            f"Try {ti.try_number} of {ti.max_tries + 1}\n"
            f"<{ti.log_url}|Logs>"
        ),
    )

def alert_sla_miss(dag, task_list, blocking_task_list, slas, blocking_tis):
    send_slack(channel="#data-alerts", text=f":hourglass: SLA missed: {task_list}")

default_args = {
    "retries": 3,
    "on_failure_callback": alert_failure,
    "sla": timedelta(hours=1),          # per task: expected completion window
}

@dag(
    ...,
    default_args=default_args,
    sla_miss_callback=alert_sla_miss,
    on_failure_callback=alert_failure,   # DAG-level: the run failed
)

An SLA in Airflow means "this task should complete within X of the DAG run's start". It does not stop or fail anything. It records a miss and fires the callback. That is a genuine limitation worth knowing: for a hard stop you need execution_timeout.

The other half of alerting is making failures self-describing:

@dag(
    ...,
    doc_md="""
    ### Daily events pipeline
    **Owner:** data-platform · **Consumers:** exec dashboard, ML features
    **On failure:** safe to clear and rerun; idempotent.
    **Escalate if:** not green by 07:00 UTC.
    """,
)

That renders in the UI, so whoever is on call at 3am has the runbook in front of them.

Custom operators and hooks

You will hit work that no provider covers. Writing an operator is short, and it is how a team stops copy-pasting the same forty lines into nine DAGs.

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
from airflow.exceptions import AirflowFailException

class RowCountCheckOperator(BaseOperator):
    """Fails if a table's row count for the interval is below a threshold."""

    template_fields = ("table", "day", "sql")     # ← makes {{ ds }} work
    ui_color = "#e8f5e9"

    def __init__(self, *, conn_id: str, table: str, day: str,
                 min_rows: int = 1, **kwargs):
        super().__init__(**kwargs)
        self.conn_id, self.table, self.day, self.min_rows = conn_id, table, day, min_rows
        self.sql = f"SELECT count(*) FROM {table} WHERE day = %s"

    def execute(self, context) -> int:
        hook = PostgresHook(postgres_conn_id=self.conn_id)
        count = hook.get_first(self.sql, parameters=(self.day,))[0]
        self.log.info("Row count for %s on %s: %s", self.table, self.day, count)
        if count < self.min_rows:
            # AirflowFailException skips remaining retries — a data problem
            # will not fix itself by trying again.
            raise AirflowFailException(
                f"{self.table} has {count} rows for {self.day}, expected >= {self.min_rows}"
            )
        return count

RowCountCheckOperator(
    task_id="check_events",
    conn_id="warehouse",
    table="staging.events",
    day="{{ ds }}",              # templated because `day` is in template_fields
    min_rows=1000,
)

Four details in there are the lesson:

template_fields is what makes {{ ds }} work in your operator's arguments. Forget it and you get the literal string, a common bug in first custom operators.

self.log is the task logger, so your messages appear in the task's log in the UI rather than in the worker's stdout.

AirflowFailException fails without consuming retries, which is correct for a data-quality failure: retrying will not add the missing rows. AirflowSkipException is its sibling, marking the task skipped.

Returning a value pushes it to XCom automatically.

A hook wraps a connection. Write one when you have a system with no provider:

from airflow.hooks.base import BaseHook
import requests

class InternalApiHook(BaseHook):
    conn_name_attr = "internal_conn_id"
    default_conn_name = "internal_default"
    hook_name = "Internal API"

    def __init__(self, internal_conn_id: str = default_conn_name):
        super().__init__()
        self.internal_conn_id = internal_conn_id

    def get_conn(self) -> requests.Session:
        conn = self.get_connection(self.internal_conn_id)   # credentials from Airflow
        session = requests.Session()
        session.headers["Authorization"] = f"Bearer {conn.password}"
        session.base_url = conn.host
        return session

    def fetch(self, path: str) -> dict:
        return self.get_conn().get(f"{self.get_connection(self.internal_conn_id).host}{path}").json()

Params and run configuration

Beginner's DAGs did the same thing every run. Params let a human parameterise a manual trigger, with validation and a form in the UI.

from airflow.models.param import Param

@dag(
    dag_id="reprocess_range",
    schedule=None,
    start_date=datetime(2024, 1, 1),
    params={
        "start_date": Param("2024-05-01", type="string", format="date",
                            title="Start date"),
        "tables": Param(["events"], type="array",
                        description="Which tables to reprocess"),
        "full_refresh": Param(False, type="boolean"),
        "batch_size": Param(1000, type="integer", minimum=1, maximum=100_000),
    },
)
def reprocess_range():

    @task
    def report(**context):
        p = context["params"]
        print(f"Reprocessing {p['tables']} from {p['start_date']}, full={p['full_refresh']}")

    BashOperator(
        task_id="run",
        bash_command="python reprocess.py --start {{ params.start_date }} "
                     "--batch {{ params.batch_size }}",
    )

Triggering with Trigger DAG w/ config gives you a generated form with the right widget for each type, and validation before the run starts.

airflow dags trigger reprocess_range \
    --conf '{"start_date": "2024-04-01", "tables": ["events","users"], "full_refresh": true}'

Testing DAGs

Beginner used airflow tasks test interactively. At this level tests belong in CI, and there are three distinct layers.

Layer one: does it import, and is the structure sane? This catches the majority of real breakages and runs in seconds.

import pytest
from airflow.models import DagBag

@pytest.fixture(scope="session")
def dagbag():
    return DagBag(dag_folder="dags/", include_examples=False)

def test_no_import_errors(dagbag):
    assert not dagbag.import_errors, f"Import errors: {dagbag.import_errors}"

def test_every_dag_has_required_metadata(dagbag):
    for dag_id, dag in dagbag.dags.items():
        assert dag.tags, f"{dag_id} has no tags"
        assert dag.default_args.get("retries", 0) >= 1, f"{dag_id} has no retries"
        assert dag.catchup is False, f"{dag_id} has catchup enabled"
        assert dag.dagrun_timeout, f"{dag_id} has no dagrun_timeout"
        assert dag.doc_md, f"{dag_id} has no documentation"

def test_no_cycles(dagbag):
    for dag in dagbag.dags.values():
        dag.test_cycle()

def test_parse_time_is_reasonable(dagbag):
    assert dagbag.dagbag_stats
    slow = [s for s in dagbag.dagbag_stats if s.duration.total_seconds() > 2]
    assert not slow, f"Slow-parsing files: {[s.file for s in slow]}"

Layer two: unit-test the logic, by extracting it from the operator.

# The business logic lives in a plain function...
def classify_load(row_count: int, threshold: int = 100_000) -> str:
    return "full_reload" if row_count > threshold else "incremental_load"

# ...so it is testable without Airflow at all.
def test_classify_load():
    assert classify_load(200_000) == "full_reload"
    assert classify_load(500) == "incremental_load"

Layer three: run the DAG. dag.test() executes a complete run in-process, which is the closest you get to an integration test.

from airflow.models import DagBag

def test_pipeline_runs_end_to_end():
    dag = DagBag(dag_folder="dags/", include_examples=False).get_dag("daily_events_pipeline")
    dag.test(execution_date=datetime(2024, 5, 1))     # raises on failure

name: DAGs
on: [pull_request]
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11', cache: pip }
      - run: pip install -r requirements.txt
      - run: ruff check dags/ plugins/
      - run: pytest tests/ -v                     # integrity + unit tests
      - run: python -c "from airflow.models import DagBag; \
               b=DagBag('dags/', include_examples=False); \
               assert not b.import_errors, b.import_errors"

Timetables and scheduling beyond cron

Cron cannot express "every weekday at 6am, but not on public holidays" or "the first business day of each month". A Timetable can.

from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction, Timetable
from pendulum import DateTime, Time

class BusinessDayTimetable(Timetable):
    """Runs at 06:00 on weekdays only, covering the previous business day."""

    def infer_manual_data_interval(self, run_after: DateTime) -> DataInterval:
        start = run_after.subtract(days=1).start_of("day")
        return DataInterval(start=start, end=start.add(days=1))

    def next_dagrun_info(self, *, last_automated_data_interval, restriction):
        if last_automated_data_interval is None:
            start = restriction.earliest
            if start is None:
                return None
        else:
            start = last_automated_data_interval.end

        # Skip forward to the next weekday
        while start.weekday() >= 5:
            start = start.add(days=1)

        if restriction.latest is not None and start > restriction.latest:
            return None

        return DagRunInfo.interval(
            start=start,
            end=start.add(days=1),
        )

@dag(schedule=BusinessDayTimetable(), start_date=datetime(2024, 1, 1), catchup=False)
def weekday_report():
    ...

Before writing one, check the built-ins. They cover more than people expect:

CronTriggerTimetable deserves a note because it resolves a real confusion: a plain cron schedule fires after the interval it covers, while CronTriggerTimetable("0 6 * * *") fires at 06:00 and treats that instant as the logical date. If you have ever wanted "run at 6am and call it today's run", that is the one.

from airflow.timetables.trigger import CronTriggerTimetable

@dag(schedule=CronTriggerTimetable("0 6 * * *", timezone="Europe/Berlin"), ...)

XCom backends and passing real data

Beginner's rule was "XCom for metadata, storage for data". At this level you can make that automatic with a custom XCom backend, so large returns go to object storage transparently.

import json, uuid
from typing import Any
from airflow.models.xcom import BaseXCom
from airflow.providers.amazon.aws.hooks.s3 import S3Hook

BUCKET = "airflow-xcom"
THRESHOLD = 8 * 1024          # anything larger goes to S3

class S3XComBackend(BaseXCom):
    @staticmethod
    def serialize_value(value: Any, **kwargs):
        payload = json.dumps(value).encode()
        if len(payload) < THRESHOLD:
            return BaseXCom.serialize_value(value)

        key = f"xcom/{uuid.uuid4()}.json"
        S3Hook().load_bytes(payload, key=key, bucket_name=BUCKET, replace=True)
        return BaseXCom.serialize_value(f"s3://{BUCKET}/{key}")

    @staticmethod
    def deserialize_value(result):
        value = BaseXCom.deserialize_value(result)
        if isinstance(value, str) and value.startswith(f"s3://{BUCKET}/xcom/"):
            key = value.split(f"{BUCKET}/", 1)[1]
            return json.loads(S3Hook().read_key(key, BUCKET))
        return value

[core]
xcom_backend = plugins.xcom.s3_backend.S3XComBackend

Small values stay in the database and remain visible in the UI; large ones move to S3 with only a pointer stored. DAG code is unchanged.

The alternative, and often the better one, is explicitness:

@task
def extract() -> str:
    df = fetch_data()
    path = f"s3://staging/events/{{{{ ds }}}}/raw.parquet"
    df.to_parquet(path)
    return path                        # a path, deliberately

@task
def transform(path: str) -> str:
    df = pd.read_parquet(path)
    ...

The explicit version is more code, and it makes the data flow visible in the DAG, which reviewers appreciate. A backend is better when many DAGs would otherwise each reimplement the same pattern.

Setup, teardown, and resource lifecycles

A DAG that provisions a cluster, uses it, and tears it down has a problem: if the middle task fails, the teardown must still run, and TriggerRule.ALL_DONE on a cleanup task is a blunt approximation.

Airflow has first-class setup and teardown tasks:

@task
def create_cluster() -> str:
    return provision_emr()          # returns a cluster id

@task
def run_job(cluster_id: str):
    submit_step(cluster_id)

@task
def destroy_cluster(cluster_id: str):
    terminate_emr(cluster_id)

cluster = create_cluster()
job = run_job(cluster)
teardown = destroy_cluster(cluster)

cluster >> job >> teardown
cluster.as_setup() >> teardown.as_teardown()      # declares the relationship

Or more concisely with the decorator arguments:

@task(multiple_outputs=False)
def create_cluster() -> str: ...

@task
def destroy_cluster(cluster_id: str): ...

with create_cluster() as cluster:          # setup/teardown context
    run_job(cluster)

The behaviour that matters: a teardown failure does not fail the DAG run unless you ask it to. That is usually right. If the job succeeded and only the cluster termination was flaky, the pipeline's output is still valid, and you want an alert rather than a red run that masks a success.

Putting it all together

Everything from this level in one DAG. Nothing here is new. Read it as a whole and you should be able to justify every line.

"""
### Events platform pipeline
**Owner:** data-platform · **Consumers:** exec dashboard, ML features
**On failure:** idempotent — safe to clear and rerun.
**Escalate if:** not green by 07:00 UTC.
"""
from datetime import datetime, timedelta

from airflow.datasets import Dataset
from airflow.decorators import dag, task, task_group
from airflow.models.param import Param
from airflow.operators.empty import EmptyOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.timetables.trigger import CronTriggerTimetable
from airflow.utils.trigger_rule import TriggerRule

EVENTS_TABLE = Dataset("postgres://warehouse/analytics.daily_events")

def alert(context):
    ti = context["task_instance"]
    print(f"ALERT {ti.dag_id}.{ti.task_id} failed — {ti.log_url}")

default_args = {
    "owner": "data-platform",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "execution_timeout": timedelta(minutes=30),
    "on_failure_callback": alert,
    "sla": timedelta(hours=2),
}

@dag(
    dag_id="events_platform",
    # Fires AT 02:00 local and calls that instant the logical date.
    schedule=CronTriggerTimetable("0 2 * * *", timezone="Europe/Berlin"),
    start_date=datetime(2024, 1, 1),
    catchup=False,
    default_args=default_args,
    dagrun_timeout=timedelta(hours=4),
    max_active_runs=1,                     # one writer per table
    max_active_tasks=12,                   # do not starve other DAGs
    tags=["production", "events"],
    doc_md=__doc__,
    params={
        "full_refresh": Param(False, type="boolean", title="Full refresh"),
        "batch_size": Param(5000, type="integer", minimum=100, maximum=50_000),
    },
    on_failure_callback=alert,
)
def events_platform():

    start = EmptyOperator(task_id="start")

    # Deferrable: releases the worker slot entirely while waiting.
    wait = S3KeySensor(
        task_id="wait_for_landing",
        bucket_name="raw-data",
        bucket_key="events/{{ ds }}/_SUCCESS",
        deferrable=True,
        poke_interval=60,
        timeout=60 * 60 * 3,
    )

    @task
    def list_partitions() -> list[str]:
        """Cheap listing at run time — never at parse time."""
        return S3Hook().list_keys(bucket_name="raw-data", prefix="events/{{ ds }}/part-")

    @task(pool="warehouse_writes", pool_slots=1, priority_weight=10)
    def load_partition(key: str) -> int:
        """Pooled: the warehouse accepts a bounded number of writers."""
        return copy_into_staging(key)

    @task_group(group_id="quality")
    def quality(counts: list[int]):
        @task
        def check_rows(c: list[int]):
            if sum(c) == 0:
                raise ValueError("No rows landed for {{ ds }}")
            return sum(c)

        @task
        def check_freshness():
            ...

        check_rows(counts) >> check_freshness()

    # Idempotent SQL: DELETE the partition, then INSERT. Marks the dataset.
    rollup = SQLExecuteQueryOperator(
        task_id="rollup",
        conn_id="warehouse",
        sql="sql/daily_rollup.sql",
        params={"lookback_days": 7},
        outlets=[EVENTS_TABLE],            # downstream DAGs are scheduled by this
    )

    cleanup = EmptyOperator(task_id="cleanup", trigger_rule=TriggerRule.ALL_DONE)
    end = EmptyOperator(task_id="end")

    counts = load_partition.expand(key=list_partitions())    # N mapped instances
    start >> wait >> counts
    quality(counts) >> rollup >> cleanup >> end

events_platform()

"""Scheduled by data, not by the clock."""
from airflow.datasets import Dataset

EVENTS_TABLE = Dataset("postgres://warehouse/analytics.daily_events")

@dag(dag_id="features_platform", schedule=[EVENTS_TABLE], start_date=..., catchup=False)
def features_platform():
    ...

Twelve decisions in there are the whole lesson of this level:

Where you are now

You can choose an executor deliberately, map tasks dynamically at run time, replace sensors with deferrable operators, schedule on data rather than time, organise large DAGs with TaskGroups and factories, protect shared systems with pools and priority, alert in a way people act on, write custom operators and hooks, parameterise runs with validation, test DAGs in CI, express schedules cron cannot, move large XCom values out of the database, and manage expensive resources with setup and teardown.

Senior takes every one of these further, with a security, scale, or ownership dimension. Who can see and trigger which DAG, and how multi-tenancy works. Secrets backends so credentials never live in the metadata database at all. Scaling the scheduler, the database, and the executor, and knowing which one is your bottleneck. Cost modelling across executors and idle capacity. Upgrade and migration strategy without a maintenance window nobody will grant you. Observability, SLOs, and the metrics that predict an incident. Incident playbooks for a stuck scheduler, a poisoned queue, a bloated metadata database, and a backfill that took down production. Running Airflow as a platform for many teams. Where Airflow stops and a streaming system, a warehouse scheduler, or an in-database tool begins.

Part two of three. A cumulative review of Beginner and Mid-level material, organised by topic rather than by level, in about thirty-five minutes. Fast review first, common questions at the end. Senior reviews all three.

Foundations, in one screen

Airflow schedules and monitors workflows defined as Python DAGs. It is an orchestrator, not a compute engine: a task should tell Spark or the warehouse to do the work.

Airflow schedules intervals, not moments. A @daily run for 1 May starts just after midnight on 2 May; the logical date is 2024-05-01. Use {{ ds }}, never now().

The scheduler executes every DAG file repeatedly, so top-level code slows everything. The metadata database is the source of truth. XCom is serialised into it, so pass paths not payloads. Credentials live in a Connection. Idempotence (delete the partition, then write) is what makes retries and backfills safe.

Executors

The trade-off is startup latency versus isolation: Celery workers are already running so a task starts in under a second, but they share an environment and you pay for idle capacity. Kubernetes gives per-task images and resources and scales to zero, at 10–30 seconds of pod startup per task.

Dynamic task mapping

@task
def list_files() -> list[str]: ...

@task
def process(key: str) -> int: ...

counts = process.expand(key=list_files())       # N instances at run time
summarise(counts)                                # receives the list of results

process.partial(bucket="raw").expand(key=keys)          # fixed + varying
process.expand_kwargs([{"k": "a"}, {"k": "b"}])          # explicit pairs

Deferrable operators and the triggerer

deferrable=True works on more than sensors. A Spark submit or a BigQuery job can defer while the external system works. The triggerer runs an asyncio loop, so trigger code must be async; a blocking call in a trigger stalls every other deferred task in that process.

Datasets and data-aware scheduling

events = Dataset("postgres://warehouse/analytics.daily_events")

@task(outlets=[events])          # producer declares it updates the dataset
def load(): ...

@dag(schedule=[events], ...)     # consumer is scheduled BY the dataset
def features(): ...

The URI is just a string Airflow does not validate, so establish a naming convention early, because inconsistent URIs mean two DAGs that look connected and are not. DatasetOrTimeSchedule combines both, and dataset expressions support & and |.

TaskGroups, factories, setup/teardown

with TaskGroup(group_id="extract") as extract:      # task_id becomes extract.orders
    for src in ["orders", "users"]:
        BashOperator(task_id=src, bash_command=f"python extract.py --src {src}")

A DAG factory generates similar DAGs from config, and must read a local file, never a database or API, because it runs on every parse.

cluster = create_cluster()
cluster.as_setup() >> destroy_cluster(cluster).as_teardown()

Pools, priority, and queues

airflow pools set legacy_db 5 "Legacy Oracle: 5 concurrent connections"

SQLExecuteQueryOperator(..., pool="legacy_db", pool_slots=1, priority_weight=10)

priority_weight defaults to downstream, so a task blocking a long chain naturally outranks a leaf. A pool's size should come from the thing it protects: a connection limit, an API rate limit, a GPU count.

SLAs, callbacks, and alerting

default_args = {"retries": 3, "on_failure_callback": alert, "sla": timedelta(hours=1)}

@dag(..., default_args=default_args, sla_miss_callback=on_sla_miss, on_failure_callback=alert)

An SLA does not stop or fail anything. It records a miss and fires the callback. For a hard stop you need execution_timeout. Alert on the DAG run failing and on SLA misses for consumer-facing pipelines; always include ti.log_url, and put the runbook in doc_md so it renders in the UI.

Custom operators and hooks

class RowCountCheckOperator(BaseOperator):
    template_fields = ("table", "day")        # ← without this, {{ ds }} stays literal

    def execute(self, context) -> int:
        count = PostgresHook(self.conn_id).get_first(self.sql, (self.day,))[0]
        self.log.info("count=%s", count)      # appears in the task log
        if count < self.min_rows:
            raise AirflowFailException("too few rows")   # no retries consumed
        return count                          # pushed to XCom

Params and run configuration

params={
    "start_date": Param("2024-05-01", type="string", format="date"),
    "batch_size": Param(1000, type="integer", minimum=1, maximum=100_000),
}

Prefer params over raw conf: a typo in a conf key is a KeyError halfway through a run at 3am, while params fails before the run starts.

Testing and CI

def test_no_import_errors(dagbag):
    assert not dagbag.import_errors

def test_standards(dagbag):
    for dag_id, dag in dagbag.dags.items():
        assert dag.tags and dag.dagrun_timeout and dag.doc_md
        assert dag.default_args.get("retries", 0) >= 1
        assert dag.catchup is False

The integrity test is the one that earns its keep: seconds to run, no database, and it catches the most common and most damaging breakage, a DAG file the scheduler cannot parse.

Timetables

The distinction worth knowing: a plain cron schedule fires after the interval it covers, while CronTriggerTimetable("0 6 * * *") fires at 06:00 and calls that instant the logical date. Set an explicit timezone once humans depend on the schedule, or the report slips an hour twice a year.

XCom backends

[core]
xcom_backend = plugins.xcom.s3_backend.S3XComBackend

A custom backend serialises small values to the database as usual and pushes large ones to object storage, storing only a pointer, with DAG code unchanged. The catch: airflow db clean removes XCom rows and knows nothing about the objects they point at, so a bucket lifecycle rule is now your responsibility.

Common interview questions

1. Celery or Kubernetes executor: how do you choose?Startup latency versus isolation. Celery workers are already running, so a task starts in under a second and thousands of short tasks are cheap, but they share an environment and you pay for idle capacity. Kubernetes gives per-task images, resources, and full isolation and scales to zero, at 10–30 seconds of pod startup per task. Many short tasks favour Celery; few heavy ones favour Kubernetes; CeleryKubernetesExecutor routes by queue.
2. Tasks are stuck in queued. Walk me through it.Check the limits in order: cluster parallelism, then the DAG's max_active_tasks, then pool slots with airflow pools list, then whether the tasks name a queue no running worker consumes. That last case looks identical to capacity exhaustion and adding workers does not fix it.
3. What is dynamic task mapping and when is it wrong?.expand() creates one task instance per element of a run-time list, each independently retryable. It is wrong when the per-item work is small: ten thousand mapped tasks is ten thousand scheduler decisions, database rows, and possibly pods, for a few minutes of real work. Map when items are substantial; batch when they are not.
4. What does expand with two arguments do?A cross product, not a zip. Two lists of three produce nine instances. Use expand_kwargs with a list of dicts for pairs.
5. What is a deferrable operator and why does it matter?The task starts, discovers it must wait, and defers, releasing its worker slot entirely and handing a lightweight trigger to the triggerer, which runs thousands in one asyncio loop. It turns a waiting task from something that costs a worker slot into something that costs almost nothing, which is why sensors in poke mode are now a fallback rather than the default.
6. What happens if the triggerer is not running?Deferred tasks sit in deferred state indefinitely and nothing alerts loudly. It is a required component once you use deferrable=True anywhere.
7. Explain data-aware scheduling.A producer task declares outlets=[Dataset(uri)]; a consumer DAG sets schedule=[Dataset(uri)] and runs when every dataset in that list has been updated since its last run. No sensor, no polling, no shared schedule assumption. The URI is an uninterpreted string, so a naming convention matters.
8. Datasets, ExternalTaskSensor, or TriggerDagRunOperator?Datasets, in almost every new case. The sensor couples both DAGs to a shared interval and costs a waiting task; the trigger operator makes the producer know about every consumer. Datasets couple only on a URI and cost nothing while waiting.
9. Why must a DAG factory read config from a file rather than a database?The factory runs on every parse cycle. A database query or API call there means every parse depends on that system being up and fast, so when it is slow the entire scheduler is slow. Sync the config to a file on a schedule and read the file.
10. What does a pool protect that max_active_tasks does not?An external system, across every DAG. max_active_tasks limits one DAG; a legacy database accepting five connections needs a limit that ten different DAGs all respect, and that is a pool. Its size should come from the resource's real limit.
11. What does an SLA in Airflow do?Records a miss and fires sla_miss_callback. It does not stop, fail, or kill anything, which surprises people. For a hard stop you need execution_timeout.
12. When does on_failure_callback fire?After retries are exhausted, not on each attempt. That is what makes it usable for alerting: a callback per attempt on a task with five retries produces five pages for one incident.
13. Your custom operator shows a literal {{ ds }}. Why?The field is not in template_fields. Only declared fields are rendered, and this is the most common bug in a first custom operator.
14. Which exception fails a task without retrying?AirflowFailException, correct for a data-quality failure, because retrying will not add the missing rows. AirflowSkipException marks it skipped instead.
15. params or dag_run.conf?params, because it gives you defaults, types, bounds, and a generated UI form, and it validates before the run starts. Raw conf is an untyped dict where a typo becomes a KeyError mid-run. Use conf only for programmatic triggers from code you control.
16. What is the single most valuable Airflow test?A DagBag integrity test: no import errors, no cycles, and your own standards asserted, so retries set, tags present, catchup=False, dagrun_timeout set. It runs in seconds, needs no database, and catches the breakage that stops the scheduler.
17. How do you enforce conventions across a team's DAGs?Assert them in the integrity test and make it a required CI check. A convention in a document is advisory; a convention that fails the build is a convention.
18. Cron string versus CronTriggerTimetable?The same expression, different semantics. A plain cron schedule fires after the interval it covers, so the logical date is the previous period. CronTriggerTimetable("0 6 * * *") fires at 06:00 and treats that instant as the logical date, which is what people usually mean by "run at 6am".
19. Why set an explicit timezone on a schedule?A UTC schedule drifts an hour relative to local business hours twice a year. Both CronTriggerTimetable and the DAG's timezone accept a real zone and handle the daylight-saving transitions, which is the fix for "the report was an hour late all summer".
20. What does a custom XCom backend make your problem?Cleanup. airflow db clean deletes XCom rows and knows nothing about the objects those rows point at, so without a bucket lifecycle rule you accumulate orphans forever.
21. Why should a teardown failure not fail the DAG run?Because the pipeline's output is still valid. If the job succeeded and only cluster termination was flaky, a red run masks a real success. You want an alert on the teardown, not a failed pipeline. That is the main behavioural difference from an ALL_DONE cleanup task.

Final self-test

• Give the Celery/Kubernetes trade-off in one sentence.
• Name the four concurrency limits and their scopes, plus the queue-routing trap.
• Say what expand with two lists produces, and when mapping is the wrong tool.
• Explain what a deferred task costs, and what happens with no triggerer.
• Describe data-aware scheduling and why it beats the two alternatives.
• Say why a DAG factory must not query a database.
• Name what a pool protects that max_active_tasks does not.
• Say what an SLA does not do, and when on_failure_callback fires.
• Give the reason a custom operator shows a literal {{ ds }}.
• Name the exception that fails without retrying, and when to use it.
• Say why params beats dag_run.conf.
• Name the cheapest high-value test and what it asserts.
• Explain cron versus CronTriggerTimetable in terms of logical date.
• Say what a custom XCom backend makes your responsibility.

Part two of three. At this level your DAGs run and your schedules are correct, so the problems change character. They are no longer "why did nothing happen" but "why is the scheduler slow", "why do twenty tasks sit in queued", and "why does this work on my laptop and not on the cluster". Start with the error table, then the practices and practice cards underneath it.

Common errors at this level

Cumulative: Beginner's errors still apply, and these are the ones that appear once things work.

The practices that pay off most

Practice cards

1. Find which limit bindsLaunch twenty parallel sleep 30 tasks and count how many run. Then lower max_active_tasks to three and repeat. Knowing your own ceiling turns "stuck in queued" into a five-minute diagnosis.
2. Queue something foreverGive a task queue="does-not-exist" and watch it sit in queued with no error anywhere. This is the hardest concurrency problem to diagnose without having seen it.
3. Meet the cross productRun expand(a=[1,2,3], b=[4,5,6]) and count the instances. Nine, not three. Then fix it with expand_kwargs.
4. Prove deferral is freeRun twenty deferrable waits and confirm zero worker slots are busy. Then run twenty poke sensors and watch the pool saturate.
5. Break the triggererStop it while deferred tasks are waiting. Confirm they hang indefinitely and nothing alerts.
6. Wire a datasetBuild a producer and a consumer, trigger the producer manually, and watch the consumer start by itself. Then change one character of the URI and watch it stop working.
7. Cap a shared resourceCreate a two-slot pool, queue six tasks into it, and confirm only two run regardless of cluster capacity. Then set `priority_weight` on two and watch the order change.
8. Make CI catch a real bugAdd the DagBag integrity test with assertions about retries and timeouts. Run it against your existing DAGs. It will fail on at least one.
9. Measure a parse-time regressionAdd a top-level import of pandas to one DAG and watch DAG Processing. Then add the budget assertion to CI and confirm it fails.

Diagnosing "stuck in queued"

This is the most common mid-level problem and it has exactly five causes. Check them in this order.

1. Cluster parallelismairflow config get-value core parallelism. If every DAG's tasks are queued, this is the first suspect.
2. The DAG's max_active_tasksIf one DAG is slow and others are fine, it is this. Visible on the DAG's details page.
3. Pool slotsairflow pools list shows used against total. A full pool queues only its own tasks, which is a useful signal.
4. Worker capacityAre workers running, and are they saturated? On Celery, check worker_concurrency and the actual replica count.
5. Queue routingDoes the task name a queue that no running worker consumes? This looks identical to capacity exhaustion and adding workers does not fix it.

airflow config get-value core parallelism
airflow pools list
airflow celery inspect active_queues          # which queues are consumed
airflow tasks states-for-dag-run my_dag <run_id>

Keeping the scheduler fast

Three measurements, and the fixes that follow from them.

airflow dags report                    # parse duration per file — start here
airflow dags list-import-errors        # a broken DAG right now

# Same syntax as .gitignore — excluded from parsing entirely
experiments/
_scratch.py

from airflow.models import DagBag
import sys

bag = DagBag("dags/", include_examples=False)
slow = [s for s in bag.dagbag_stats if s.duration.total_seconds() > 2]
if slow:
    print("Slow-parsing files:", [(s.file, s.duration) for s in slow])
    sys.exit(1)

Alerting people read

def alert(context):
    ti = context["task_instance"]
    post_to_slack(
        channel="#data-alerts",
        text=(
            f":red_circle: *{ti.dag_id}.{ti.task_id}* failed\n"
            f"Logical date: {context['ds']} · try {ti.try_number}/{ti.max_tries + 1}\n"
            f"<{ti.log_url}|Logs>"
        ),
    )

Testing that catches real breakage

import pytest
from airflow.models import DagBag

@pytest.fixture(scope="session")
def dagbag():
    return DagBag(dag_folder="dags/", include_examples=False)

def test_no_import_errors(dagbag):
    assert not dagbag.import_errors, dagbag.import_errors

def test_standards(dagbag):
    for dag_id, dag in dagbag.dags.items():
        assert dag.tags, f"{dag_id}: no tags"
        assert dag.catchup is False, f"{dag_id}: catchup enabled"
        assert dag.dagrun_timeout, f"{dag_id}: no dagrun_timeout"
        assert dag.doc_md, f"{dag_id}: no documentation"
        assert dag.default_args.get("retries", 0) >= 1, f"{dag_id}: no retries"
        for task in dag.tasks:
            assert task.execution_timeout, f"{dag_id}.{task.task_id}: no timeout"

def test_no_cycles(dagbag):
    for dag in dagbag.dags.values():
        dag.test_cycle()

"Works locally, fails on the cluster"

Five differences, roughly in the order they are the answer.

# On the cluster, not your laptop
airflow info
airflow connections get warehouse
airflow providers list
airflow tasks test my_dag my_task 2024-05-01   # run it where it fails

Habits worth adopting now

Convert every sensor to deferrable=True. It is usually a one-word change and it is the single largest capacity and cost win available.

Put a pool in front of every shared system. Sized from the resource's real limit, with the reason in the pool description.

Prefer datasets to ExternalTaskSensor. Nothing is held while waiting, and neither DAG needs to know the other's schedule.

Batch before you map. Mapping is for substantial, independently retryable items. Ten thousand two-second tasks is a scheduler problem, not a parallelism win.

Assert the parse-time budget in CI. Degradation arrives one import at a time and is invisible until it is severe.

Extract logic out of operators. A plain function is unit-testable; an operator's execute is not.

Schedule db clean. The metadata database is the most common scaling ceiling and retention is the fix.

# A mid-level default_args worth copying
DEFAULTS = {
    "owner": "your-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(minutes=30),
    "on_failure_callback": alert,          # actionable, with a log link
    "sla": timedelta(hours=2),             # records a miss; does not stop anything
}

Senior tips go further: the hardening pass every deployment should get, verifying controls rather than trusting them, cost governance, incident playbooks for a stuck scheduler and a backfill that took production down, running Airflow as a platform, and where Airflow stops.

This is part three of three. It closes the series by taking every topic from Beginner and Mid one level further (each one now has a security, scale, or ownership dimension) and adds the work you own when Airflow is your responsibility rather than your tool.

Up to this point the questions have been about making a DAG run correctly. From here they are different: who can see and trigger which pipeline, where the credentials live, which component breaks first under load, what the whole thing costs, how you upgrade it without a maintenance window nobody will grant you, and what you do at 3am when the scheduler stops scheduling.

Where this picks up

I am starting with the security model, because every other decision in this track depends on it.

The security model

Airflow's security surface is larger than people expect, and the reason is a single fact: a DAG file is arbitrary Python that Airflow executes with the scheduler's and workers' privileges.

Three properties make this sharper than it first appears:

Anyone who can write a DAG file has remote code execution on your cluster. There is no sandbox. A DAG file could read every connection in the metadata database, exfiltrate them, and delete its own logs. DAG-folder write access is therefore equivalent to administrative access, and the control is code review plus a deployment pipeline, not Airflow configuration.

Logs leak secrets by default. Airflow masks values it knows are sensitive, based on connection fields and a configured list of key names. A credential you fetched yourself and printed is not masked. This is the most common real-world Airflow security incident, and it is entirely preventable.

[core]
hide_sensitive_var_conn_fields = True
sensitive_var_conn_names = api_token, private_key, client_secret

The UI shows more than people assume. By default an authenticated user can read every DAG's source, every log, and the list of connections. Variables' values are masked when the name looks sensitive, which is a heuristic, not a boundary.

RBAC and multi-tenancy

Airflow ships role-based access control. Understanding what it does and does not give you is the difference between a plan and a hope.

Per-DAG permissions are the mechanism that makes team separation possible:

# A role scoped to one team's DAGs
airflow roles create team_billing
airflow roles add-perms team_billing \
    -r can_read -v DAG:billing_ingest \
    -r can_edit -v DAG:billing_ingest

# Or declare access at the DAG level
@dag(
    dag_id="billing_ingest",
    access_control={"team_billing": {"can_read", "can_edit"}},
    ...
)

So genuine multi-tenancy needs infrastructure, not configuration:

Secrets backends

The strongest fix for the previous section's problem is to take credentials out of Airflow entirely.

[secrets]
backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
backend_kwargs = {
    "connections_path": "airflow/connections",
    "variables_path": "airflow/variables",
    "mount_point": "secret",
    "url": "https://vault.internal:8200",
    "auth_type": "kubernetes",
    "kubernetes_role": "airflow"
}

Airflow resolves a conn_id through the backend chain: the secrets backend first, then environment variables, then the metadata database. So conn_id="warehouse" becomes a lookup at secret/airflow/connections/warehouse in Vault, with nothing stored locally.

Three consequences worth stating:

Rotation stops being a project. A credential rotated in Vault is picked up on the next task run with no Airflow change and no restart.

Reads become auditable. The secret store logs which identity read which secret when, which is the evidence an auditor asks for and the metadata database cannot provide.

Per-team scoping becomes real. Workers in the billing namespace authenticate with a service account that can only read secret/airflow/connections/billing/*. A DAG that asks for another team's connection gets a denial from Vault rather than a password from your database.

Scaling: which component breaks first

Mid taught you the components. At scale you need to know which one saturates, because the symptom is always "Airflow is slow" and the cause is one of four different things.

[scheduler]
# Parsing
parsing_processes = 4                    # parallel DAG file processors
min_file_process_interval = 60           # re-parse each file at most this often
dag_dir_list_interval = 120              # rescan the folder this often
file_parsing_sort_mode = modified_time   # prioritise recently changed files

# Loop throughput
max_tis_per_query = 512                  # task instances examined per query
scheduler_heartbeat_sec = 5
use_row_level_locking = True             # required for multiple schedulers

[database]
sql_alchemy_pool_size = 10
sql_alchemy_max_overflow = 20
sql_alchemy_pool_recycle = 1800

Three scaling facts that matter more than the rest:

Multiple schedulers are active-active. Since Airflow 2 you run two or three schedulers behind row-level locking and they share the work. This is the answer to both throughput and availability, and it needs a database that supports SELECT … FOR UPDATE SKIP LOCKED: Postgres 12+ or MySQL 8+.

The metadata database is almost always the real ceiling. Every heartbeat, state change, XCom write, and UI page is a query. A cluster running fifty thousand task instances a day generates a large amount of write traffic, and the fix is usually retention rather than a bigger machine.

airflow db clean --clean-before-timestamp 2024-01-01
airflow db clean --clean-before-timestamp 2024-01-01 --tables task_instance,dag_run,log,xcom

Raising min_file_process_interval is the cheapest parsing win. The default re-parses every file every thirty seconds. Sixty or a hundred and twenty seconds is almost always acceptable. New DAGs appear a minute later, and the scheduler does a fraction of the work.

Cost

Airflow's cost is rarely the scheduler. It is idle capacity, the database, and the compute your tasks trigger.

Five levers, roughly in order of return:

1. Deferrable operators everywhereA sensor in poke mode on a GPU-capable worker is the most expensive way to wait ever invented. Deferring moves the wait to one small async process.
2. Right-size the executor to the workloadMany short tasks favour Celery, because pod startup dominates. Few heavy tasks favour Kubernetes, because idle workers dominate. A CeleryKubernetesExecutor with queue routing gets both.
3. Scale workers on queue depthKEDA on Celery queue length, or Kubernetes with scale-to-zero. Sizing for peak and paying overnight is the default and the most common waste.
4. Retention on everythingdb clean for the database, a lifecycle rule for remote logs. Both grow silently and neither is ever read after a month.
5. Batch instead of mapping when items are smallTen thousand mapped tasks each doing two seconds of work is ten thousand schedules, rows, and possibly pods, for twenty seconds of real work.

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
spec:
  scaleTargetRef:
    name: airflow-worker
  minReplicaCount: 1
  maxReplicaCount: 20
  triggers:
    - type: postgresql
      metadata:
        query: >-
          SELECT ceil(count(*)::decimal / 16)
          FROM task_instance
          WHERE state = 'queued' OR (state = 'running' AND queue = 'default')
        targetQueryValue: "1"

Observability and SLOs

Mid added alerting on failures. At platform scale you need to know Airflow is healthy before a pipeline misses, which means metrics rather than alerts.

[metrics]
statsd_on = True
statsd_host = statsd.monitoring
statsd_port = 8125
statsd_prefix = airflow

The metrics that predict incidents:

Then define SLOs on the things consumers care about:

That last distinction is the one that makes a platform team's life livable: separate platform SLOs from pipeline SLOs. "Runs start on time" is the platform's promise. "The data is correct and complete" is the DAG owner's. Conflating them means the platform team is paged for every team's data bug.

Upgrades and migrations

An Airflow upgrade touches the database schema, the provider packages, and every DAG's imports at once. Doing it without a plan is how teams end up three major versions behind.

1. Read the release notes properlyDeprecations become removals a version later. The upgrade that breaks you is usually one you skipped, not the one you are doing.
2. Run the upgrade checkairflow db check-migrations and, on older versions, the dedicated upgrade-check tooling. Fix what it reports before touching anything.
3. Pin providers separately from coreProvider packages version independently. Upgrading a provider for a bug fix should not require an Airflow upgrade, and that is only true if you pin them separately.
4. Rehearse the migration on a copy of production dataairflow db migrate on a two-hundred-gigabyte task_instance table can take hours. Time it on a restored snapshot, not in the maintenance window.
5. Clean the database firstA smaller table migrates faster. db clean before an upgrade often turns a long outage into a short one.
6. Deploy DAG compatibility ahead of the upgradeImport paths change between versions. Where both old and new imports work, move to the new ones in advance so the upgrade itself is only infrastructure.

# Before
airflow db check-migrations
airflow db clean --clean-before-timestamp $(date -d '90 days ago' -I)
pg_dump airflow > backup-$(date -I).sql        # non-negotiable

# The upgrade
pip install "apache-airflow==2.10.*" --constraint "$CONSTRAINTS_URL"
airflow db migrate

# After
airflow dags list-import-errors
airflow jobs check --job-type SchedulerJob --hostname "$(hostname)"

apache-airflow==2.10.2
apache-airflow-providers-amazon==8.28.0
apache-airflow-providers-postgres==5.12.0
# Always install with the official constraints file for your version and Python

Deployment and DAG delivery

How DAG files reach the scheduler is an architectural choice with real consequences.

Whichever you pick, the gate is the same:

name: DAGs
on: [pull_request]
jobs:
  gate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11', cache: pip }
      - run: pip install -r requirements.txt --constraint constraints.txt
      - run: ruff check dags/ plugins/
      - name: Integrity — no import errors, and standards enforced
        run: pytest tests/test_dag_integrity.py -v
      - name: Unit tests for extracted logic
        run: pytest tests/unit -v
      - name: Parse-time budget
        run: python scripts/assert_parse_time.py --max-seconds 2

Running Airflow as a platform

Once several teams share one Airflow, the highest-leverage work stops being technical and becomes about defaults. Four decisions cover most of it.

Publish a template repository, not documentation. A working DAG with retries, timeouts, tags, doc_md, and idempotent SQL; the integrity test suite; the CI workflow. Teams inherit every good decision by cloning, and a wiki page is read once.

Publish a shared operator library as a versioned package. The row-count check, the Slack alert callback, the standard extract pattern. It is where you enforce standards: a quality gate that is one line gets adopted, and forty lines does not.

Own the contract, not the DAGs. The platform team owns scheduler health, punctuality, the shared library, and the deployment gate. DAG owners own correctness, their SLAs, and their own on-call. Writing that split down is what stops the platform team being paged for every data bug.

Measure and publish. Without numbers, "Airflow is fine" is an opinion:

Incident playbooks

Under pressure, ordering matters more than knowledge. Five shapes, with the first move for each.

The scheduler has stopped scheduling

Nothing is failing. Nothing is running. Confirm first with airflow jobs check --job-type SchedulerJob, because a dead scheduler produces silence rather than alerts.

Then work the four causes in order of likelihood: is the process alive; can it reach the metadata database; is one DAG file taking longer than dag_file_processor_timeout and blocking a processor slot; and is the database itself refusing connections because the pool is exhausted. airflow dags report names a pathological file immediately, and .airflowignore gets it out of the way while you fix it. Afterwards: the heartbeat alert, if you did not already have it.

Tasks are stuck in queued

They are scheduled but nothing picks them up, which is always a capacity or routing problem. Check in order: parallelism cluster-wide, then the DAG's max_active_tasks, then pool slots, airflow pools list shows used against total, then whether the tasks name a queue that no running worker consumes. That last one is the sneaky case, because it looks identical to capacity exhaustion and no amount of extra workers fixes it.

The metadata database is degrading

The UI times out, the scheduler loop lengthens, everything is slow. Check size and row counts first, then the connection count against the server's limit. The immediate action is db clean with a conservative cutoff; the durable fixes are scheduled retention, PgBouncer in front of a Postgres with hundreds of Airflow connections, and moving large XCom values out of the database. Resist the instinct to resize the instance first. It buys weeks and hides the cause.

A backfill took production down

A secret leaked into a task log

Rotate before anything else. The log is in object storage, in a log aggregator, possibly in a Slack alert, and in anyone's browser cache. Deleting the log line is not containment. Then find the code path that printed it, and fix the class rather than the instance: add the key name to sensitive_var_conn_names, and add a CI check that greps DAG code for printing connection or variable values. Rotating and moving on without closing the class means it happens again with a different credential.

Where Airflow stops

Naming the boundary is itself a senior signal, and it lands better than defending Airflow for every workload.

The shape that works in practice: Airflow schedules and orchestrates, the heavy work happens in a purpose-built engine, and transformation logic lives in dbt or SQL rather than in Python inside a task. A DAG whose tasks are mostly SubmitSparkJob, RunDbtModel, and WaitForBigQueryJob is a well-designed DAG. A DAG whose tasks load dataframes into worker memory is a misuse that will eventually force a rewrite.

The review checklist

Take this checklist away from this level. Run it against any DAG (your own, or one you are reviewing) and it catches nearly everything covered here before it becomes an incident.

The complete picture

The series' final artefact: every level's topics, hardened. Nothing in it is new.

"""
### Events platform pipeline
**Owner:** data-platform · **Consumers:** exec dashboard, ML features
**SLO:** green by 07:00 UTC, 99% of days.
**On failure:** idempotent — safe to clear and rerun any interval.
**Escalate:** #data-oncall if not green by 07:30 UTC.
"""
from datetime import datetime, timedelta

from airflow.datasets import Dataset
from airflow.decorators import dag, task, task_group
from airflow.models.param import Param
from airflow.operators.empty import EmptyOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.timetables.trigger import CronTriggerTimetable
from airflow.utils.trigger_rule import TriggerRule

from company_airflow.callbacks import slack_alert      # the shared library
from company_airflow.operators import RowCountCheckOperator

EVENTS = Dataset("postgres://warehouse/analytics.daily_events")

default_args = {
    "owner": "data-platform",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(minutes=30),   # no task holds a slot forever
    "on_failure_callback": slack_alert,           # actionable, with a log link
    "sla": timedelta(hours=2),
}

@dag(
    dag_id="events_platform",
    schedule=CronTriggerTimetable("0 2 * * *", timezone="Europe/Berlin"),
    start_date=datetime(2024, 1, 1),
    catchup=False,                                # backfill deliberately
    default_args=default_args,
    dagrun_timeout=timedelta(hours=4),
    max_active_runs=1,                            # one writer per table
    max_active_tasks=12,                          # do not starve other teams
    tags=["production", "events", "tier-1"],
    doc_md=__doc__,
    # Only this team may trigger or edit; others may look.
    access_control={
        "team_data_platform": {"can_read", "can_edit"},
        "Viewer": {"can_read"},
    },
    params={
        "full_refresh": Param(False, type="boolean", title="Full refresh"),
        "batch_size": Param(5000, type="integer", minimum=100, maximum=50_000),
    },
    on_failure_callback=slack_alert,
)
def events_platform():

    start = EmptyOperator(task_id="start")

    # Deferrable: zero worker slots held for up to three hours of waiting.
    wait = S3KeySensor(
        task_id="wait_for_landing",
        bucket_name="raw-data",
        bucket_key="events/{{ ds }}/_SUCCESS",
        aws_conn_id="aws_raw",          # resolved from the secrets backend
        deferrable=True,
        poke_interval=60,
        timeout=60 * 60 * 3,
    )

    @task
    def list_partitions() -> list[str]:
        """Run-time listing. Nothing at module level touches the network."""
        return S3Hook(aws_conn_id="aws_raw").list_keys(
            bucket_name="raw-data", prefix="events/{{ ds }}/part-"
        )

    @task(pool="warehouse_writes", pool_slots=1, priority_weight=10)
    def load_partition(key: str) -> int:
        """Pooled: the warehouse's connection limit is respected cluster-wide.
        Each partition retries independently."""
        return copy_into_staging(key)

    @task_group(group_id="quality")
    def quality(counts: list[int]):
        # A data problem will not fix itself: this raises AirflowFailException.
        RowCountCheckOperator(
            task_id="row_count",
            conn_id="warehouse",
            table="staging.events",
            day="{{ ds }}",
            min_rows=1000,
        )

        @task
        def totals(c: list[int]) -> int:
            return sum(c)

        totals(counts)

    # Idempotent: DELETE the partition then INSERT. Publishes the dataset.
    rollup = SQLExecuteQueryOperator(
        task_id="rollup",
        conn_id="warehouse",
        sql="sql/daily_rollup.sql",
        params={"lookback_days": 7},
        outlets=[EVENTS],                 # downstream DAGs are data-scheduled
    )

    cleanup = EmptyOperator(task_id="cleanup", trigger_rule=TriggerRule.ALL_DONE)
    end = EmptyOperator(task_id="end")

    counts = load_partition.expand(key=list_partitions())
    start >> wait >> counts
    quality(counts) >> rollup >> cleanup >> end

events_platform()

[core]
executor = CeleryKubernetesExecutor
parallelism = 128
max_active_tasks_per_dag = 16
hide_sensitive_var_conn_fields = True

[secrets]
backend = airflow.providers.hashicorp.secrets.vault.VaultBackend

[scheduler]
parsing_processes = 4
min_file_process_interval = 120        # cheapest parsing win there is
use_row_level_locking = True           # multiple active schedulers

[metrics]
statsd_on = True                       # loop duration, starving tasks, parse time

[webserver]
rbac = True
expose_config = False

# Verify, do not assume
airflow jobs check --job-type SchedulerJob      # is the scheduler alive?
airflow dags report                              # parse-time distribution
airflow dags list-import-errors                  # anything broken right now?
airflow pools list                               # used vs total slots
airflow db clean --dry-run --clean-before-timestamp $(date -d '90 days ago' -I)

Where the series leaves you

Across the three levels you have gone from a first three-task DAG to owning Airflow as a platform. The same topics carried all the way through, each time with more depth:

You should now be able to look at an Airflow deployment and see not just what it does but what it guarantees: whether a run from last quarter can be reproduced, whether a hung task can take the cluster with it, who can read which credential, which component will saturate first and what the metric for it is, what it costs and how much of that is idle, whether a broken DAG can reach production, and whether anyone finds out when the scheduler stops.

That review instinct matters more than any single configuration flag, because, as with containers and data versioning, most orchestration incidents are prevented at review time rather than at 3am.

Part three of three, and the one to read if you only read one. A cumulative review of the entire series (foundations, pipeline machinery, and the security and platform work a senior owns) organised by topic rather than by level. About fifty minutes. Fast review first, common questions at the end.

Part one: foundations

Airflow schedules and monitors workflows defined as Python DAGs. It gives you a dependency graph, retries, historical intervals, and full run history, replacing cron chains that had none of those. It is an orchestrator, not a compute engine.

Components, and their symptoms

The scheduling model

Airflow schedules intervals, not moments. A @daily run for 1 May starts just after midnight on 2 May; the logical date is 2024-05-01. Use {{ ds }}, never now(), or reruns and backfills stop being reproducible.

catchup=True with an old start_date creates every missed interval at once, hundreds of runs on unpause.

The rules that never change

The scheduler executes every DAG file repeatedly, so no queries, API calls, Variable.get(), or heavy imports at the top level. XCom is serialised into the metadata database, so pass paths not payloads. Credentials live in a Connection. Idempotence (delete the partition, then write) is what makes retries and backfills safe. A skip is neither success nor failure and it propagates, so a join after a branch needs an explicit trigger rule.

Part two: pipeline machinery

Executors and limits

Mapping, deferring, datasets

counts = process.expand(key=list_files())      # N run-time instances

expand with two arguments is a cross product; use expand_kwargs for pairs. Mapping is wrong when items are tiny; batch instead.

Datasets: a producer declares outlets=[Dataset(uri)], a consumer sets schedule=[Dataset(uri)]. No sensor, no shared interval, nothing held while waiting. The URI is an unvalidated string, so conventions matter.

Organisation and control

TaskGroups collapse a large graph with no runtime cost; task_id becomes group.task. A DAG factory must read a local file, never a database, because it runs every parse. Setup/teardown expresses resource lifecycles, and a teardown failure does not fail the run, which is right, because the output is still valid.

Pools protect an external system across all DAGs; pool_slots lets one heavy task count as several; priority_weight defaults to downstream so a task blocking a chain outranks a leaf.

Alerting, operators, params, testing

An SLA does not stop or fail anything. It records a miss. on_failure_callback fires after retries are exhausted, not per attempt. Always include ti.log_url; put the runbook in doc_md.

In a custom operator: template_fields is what makes {{ ds }} render, self.log puts messages in the task log, AirflowFailException fails without consuming retries, and a returned value goes to XCom.

params over dag_run.conf: types, bounds, a UI form, and validation before the run starts.

The DagBag integrity test is the highest-value test: seconds, no database, and it catches import errors plus any standard you assert.

Timetables and XCom backends

A plain cron schedule fires after the interval; CronTriggerTimetable("0 6 * * *") fires at 06:00 and calls that the logical date. Set an explicit timezone once humans depend on the schedule.

A custom XCom backend moves large values to object storage transparently, and makes cleaning up those objects your problem, because db clean only removes rows.

Part three: security, scale, and ownership

The security model

A DAG file is arbitrary Python executed with the scheduler's and workers' privileges. There is no sandbox.

RBAC and multi-tenancy

Real isolation needs infrastructure: Kubernetes namespaces and service accounts per team, a secrets backend with per-team paths, separate queues or node pools, plus pools. For mutually untrusting tenants the honest answer is separate deployments.

Secrets backends

Airflow resolves a conn_id through the backend chain: secrets backend, then environment variables, then the metadata database.

It is on the hot path of every task, so monitor its latency as a first-class dependency and never point it at something less available than Airflow.

Scaling: which component breaks first

Multiple schedulers are active-active via row-level locking, the answer to both throughput and availability. Raising min_file_process_interval is the cheapest parsing win. The database is almost always the real ceiling, and retention is the fix rather than size.

Cost

Worker utilisation, task-seconds over paid worker-seconds, is the most revealing metric and is usually far lower than people guess.

Observability and SLOs

Separate platform SLOs from pipeline SLOs. Punctuality and scheduler availability are the platform's promise; correctness and completeness are the DAG owner's. Conflating them means the platform team is paged for every data bug.

Upgrades and delivery

There is no downgrade path for airflow db migrate. The rollback plan is a tested backup. Clean the database first: a smaller task_instance table turns a long migration into a short one. Pin providers separately from core, and move deprecated import paths ahead of the upgrade so the upgrade itself is only infrastructure.

Git-sync plus a required, fast integrity check on the followed branch is the pragmatic middle. Git-sync from a branch anyone can push to without CI is a production deployment with no gate.

Where Airflow stops

Airflow has no streaming semantics and should not compute. A DAG of SubmitSparkJob, RunDbtModel, and WaitForBigQueryJob is well designed; a DAG loading dataframes into workers will force a rewrite.

The review checklist

Common interview questions

1. Explain Airflow to an engineer who has never used it.A platform for scheduling and monitoring workflows defined as Python DAGs. It computes the execution order from declared dependencies, schedules historical intervals rather than moments, retries failures, and records every run in a metadata database with a UI over it. It is an orchestrator: the heavy work should happen in Spark, the warehouse, or a container, not in the worker.
2. Why is DAG-folder write access a security concern?Because a DAG file is arbitrary Python that the scheduler and workers execute with their own privileges, and there is no sandbox. A DAG could read every connection in the metadata database and exfiltrate it. So write access to that folder is administrative access, and the control is code review plus a deployment pipeline, not Airflow configuration.
3. What does RBAC give you, and what does it not?It gives per-DAG read, edit, and delete permissions, separates who can trigger from who can view, restricts connection management, and integrates with SSO. It does not isolate task code between teams, does not stop one team's DAG reading another's credential, and offers nothing against a malicious DAG author.
4. Can one Airflow safely host two mutually untrusting teams?Not if both author DAGs. A DAG file runs arbitrary code, so no permission model prevents one team reading the other's data or secrets. The honest answers are separate deployments, or a model where nobody writes DAGs directly and pipelines are generated from a restricted specification.
5. What does a secrets backend buy you over the metadata database?Native rotation with no Airflow change, per-team scoping by path and IAM so a DAG asking for another team's credential is denied at the source, and an audit trail of which identity read which secret when. The cost is that it sits on the hot path of every task, so its latency and availability become yours.
6. Why does log masking sometimes fail?Airflow masks values it knows are sensitive: connection fields and configured key names. A credential you fetched into a variable yourself and printed is not on that list. This is the most common real Airflow security incident and it is entirely preventable, both by configuration and by a CI check on printing.
7. Four things are slow. Which component is it?New DAGs appearing slowly with pinned scheduler CPU is parsing. Tasks sitting in scheduled is the scheduler loop. Everything slow including the UI is the metadata database. Tasks sitting in queued is worker capacity, or a queue no worker consumes, which looks identical and is not fixed by adding workers.
8. What is almost always the real scaling ceiling?The metadata database. Every heartbeat, state change, XCom write, and UI page is a query, and without retention task_instance and log grow forever. The fix is db clean on a schedule and moving large XCom values out. Resizing the instance buys weeks and hides the cause.
9. How do you run Airflow highly available?Multiple schedulers, active-active, coordinating through row-level locking in the database, which needs Postgres 12+ or MySQL 8+. That covers throughput and availability at once. The webserver is independently replicable, and its absence does not stop pipelines.
10. Give the cheapest performance win in a busy deployment.Raising min_file_process_interval. The default re-parses every file every thirty seconds; a hundred and twenty seconds means new DAGs appear a minute later and the scheduler does a fraction of the work.
11. Where does Airflow's cost go?Rarely the scheduler. It is idle worker capacity, sized for the 2am peak and paid for all day, plus the metadata database, plus whatever compute the tasks trigger. Deferrable operators remove the worst waste, autoscaling on queue depth removes the next, and retention removes the quiet growth.
12. What single metric would you put on a dashboard?Worker utilisation: task-seconds divided by paid worker-seconds. It is usually far lower than anyone guesses, and a team that can see 12% will fix it while a team that cannot will keep adding workers.
13. What is the one alert every Airflow deployment needs?Scheduler heartbeat age. If the scheduler stops, everything stops silently: no failures, no red DAGs, because failure alerts come from tasks that never ran. Its absence is why "nothing had run since 2am and we found out at 10" is such a common story.
14. How do you split platform SLOs from pipeline SLOs?The platform promises punctuality and scheduler availability: runs start within N minutes of schedule, the scheduler heartbeats. DAG owners promise correctness, completeness, and freshness. Conflating them means the platform team is on call for every team's data bug, which does not scale and does not last.
15. Walk me through an Airflow upgrade.Read the release notes for deprecations and removals, run the migration check, clean the database so the migration is shorter, take and test a backup, deploy DAG import-path changes ahead of time where both forms work, then upgrade with the official constraints file and run db migrate. Rehearse the migration timing on a restored production snapshot, not in the window.
16. What is the rollback plan for db migrate?A tested database backup. There is no practical downgrade, the schema changes are not reversible, so a backup you have never restored is a hypothesis rather than a plan.
17. How should DAGs reach the scheduler?Baked into an image is safest: DAGs, providers, and dependencies versioned together, and rollback is a redeploy. Git-sync is fastest to iterate with but ships a broken DAG as quickly as a good one. The pragmatic middle is git-sync from a branch protected by a required, fast integrity check. Git-sync from an unprotected branch is a deployment with no gate.
18. Nothing has run for two hours and nothing has failed. What do you do?Check the scheduler first with airflow jobs check, because silence means the scheduler, not the tasks. Then: is the process alive, can it reach the database, is one DAG file exceeding dag_file_processor_timeout and blocking a processor, and is the database connection pool exhausted. airflow dags report names a pathological file immediately.
19. A backfill took production down. What now?Pause the DAG and clear the queued backfill instances so they stop competing, confirm production drains and recovers, then find what it exhausted: worker slots, a pool, database connections, or the warehouse. Restart it with max_active_runs=1, a dedicated small pool, and low priority. The durable fix is a standing backfill pool and a convention that backfills run outside business hours.
20. How do you contain a noisy neighbour on a shared Airflow?Pools for shared external systems, per-team queues or node pools for compute, `max_active_tasks` per DAG, and a parse-time budget enforced in CI. Without those, one team's careless commit is an incident for every other team, which on a shared deployment is the primary engineering requirement, not a refinement.
21. How would you roll Airflow out to six teams?A template repository with a working DAG, the integrity tests, and CI, so teams inherit every decision by cloning. A shared operator library as a versioned package, because a quality gate that is one line gets adopted and forty lines does not. A written contract splitting platform ownership from pipeline ownership. Published metrics: punctuality, parse time, utilisation, and standards adoption per team.
22. Where does Airflow stop?Sub-second and event-driven work belongs in Kafka or Flink; there are no streaming semantics here. Pure SQL transformation belongs in dbt, run from Airflow. Per-request triggering belongs in a service. Heavy computation belongs in Spark, Ray, or the warehouse; a task that loads a large dataframe into a worker is a misuse that will force a rewrite.
23. A team wants to migrate off Airflow. How do you respond?Diagnose the practices first. Usually the complaints trace to top-level code, sensors in poke mode, computation inside tasks, no retention, and no pools, all fixable in weeks. Migrating orchestrators is a multi-quarter project that faithfully reproduces every one of those mistakes in a new tool unless the practices change first.
24. What do you look for reviewing a DAG?Not what it does, but what it guarantees. Can any interval be reproduced; is every task idempotent; can a hung task take a worker with it; who can read the credentials it uses; does it protect the systems it writes to; will anyone find out when it fails; and is the heavy work happening somewhere other than the worker.

Final self-test

• Give the whole-series answer in four sentences, including "orchestrator, not compute engine".
• Explain why DAG-folder write access is administrative access.
• Say what RBAC does and does not isolate, and the honest multi-tenancy answer.
• Name what a secrets backend buys, and the cost it introduces.
• Say why log masking fails, and the two fixes.
• Map four "slow" symptoms to four components, including the queue-routing trap.
• Name the usual real scaling ceiling and the correct first fix.
• Say how Airflow runs highly available, and what the database must support.
• Give the cheapest parsing win, and the biggest cost lever.
• Name the one metric and the one alert every deployment needs.
• Split platform SLOs from pipeline SLOs in one sentence each.
• Give the rollback plan for db migrate, and why the backup must be tested.
• Say what to check first when nothing has run and nothing has failed.
• Name four noisy-neighbour containment mechanisms.
• Say where Airflow stops, and how you answer "should we migrate off it".
• Run the review checklist from memory, and name which rows a machine can check.

Part three of three. The problems at this level are rarely about a DAG. They are about a scheduler sixty people depend on, a metadata database nobody has cleaned in two years, a credential in a log, and a backfill that took production down. Start with the error table, then the practices, verification, and playbooks underneath it.

Common errors at this level

Cumulative. Everything from Beginner and Mid still applies. These cause incidents rather than failed tasks.

The practices that pay off most

Practice cards

1. Kill the scheduler and time your detectionStop it in a test environment and measure how long until monitoring tells you, rather than you noticing. Most deployments are silent.
2. Prove a DAG can read anythingWrite a DAG under one team's ownership that reads a connection belonging to another. It works, and that result is the whole argument for a secrets backend.
3. Check the masking gapPrint a connection password directly, then fetch the same secret into a variable with an innocuous name and print that. Compare what the log masks.
4. Time a real migrationRestore a production snapshot into a scratch instance and run airflow db migrate against it. That number is your maintenance window.
5. Measure what db clean would removeDry-run a ninety-day cutoff and read the row counts. It is usually enough to schedule the real thing immediately.
6. Calculate worker utilisationTotal task duration last week divided by total worker uptime paid for. Put the number somewhere visible.
7. Cause the queue-routing hangGive a task a queue no worker consumes and practise diagnosing it from the UI alone. It is the hardest of these to recognise cold.
8. Reproduce a historical runClear an interval from three months ago and confirm the output is identical rather than duplicated. If it is not, idempotence is broken somewhere.
9. Run the checklist on someone else's DAGThe review checklist below, against a pipeline you did not write. It will find something.

The hardening pass every deployment should get

[core]
executor = CeleryKubernetesExecutor
parallelism = 128
max_active_tasks_per_dag = 16
hide_sensitive_var_conn_fields = True
sensitive_var_conn_names = api_token, private_key, client_secret

[secrets]
backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
# Credentials never live in the metadata database.

[scheduler]
parsing_processes = 4
min_file_process_interval = 120        # the cheapest parsing win there is
dag_file_processor_timeout = 60        # a pathological file cannot block forever
use_row_level_locking = True           # multiple active-active schedulers

[database]
sql_alchemy_pool_size = 10
sql_alchemy_max_overflow = 20
sql_alchemy_pool_recycle = 1800

[metrics]
statsd_on = True                       # loop duration, starving tasks, parse time

[webserver]
rbac = True
expose_config = False                  # do not show config in the UI

@dag(
    ...,
    # Only this team may trigger or edit; others may look.
    access_control={
        "team_data_platform": {"can_read", "can_edit"},
        "Viewer": {"can_read"},
    },
)

Verifying, not assuming

# Is the scheduler alive, and how stale is its heartbeat?
airflow jobs check --job-type SchedulerJob --allow-multiple

# Does the secrets backend actually serve this connection?
airflow connections get warehouse          # then delete it from the UI and retry

# What would retention remove?
airflow db clean --dry-run --clean-before-timestamp $(date -d '90 days ago' -I)

# Which files are pathological?
airflow dags report | sort -k3 -h | tail -10
airflow dags list-import-errors

# Are pools doing anything?
airflow pools list

# Which queues are actually consumed?
airflow celery inspect active_queues

# Does log masking work for the value you care about?
airflow tasks test my_dag print_secret 2024-05-01 | grep -i token

Automate the mechanical checks so they do not depend on anyone remembering:

- name: No import errors, and standards enforced
  run: pytest tests/test_dag_integrity.py -v

- name: Parse-time budget
  run: python scripts/assert_parse_time.py --max-seconds 2

- name: No secrets printed in DAG code
  run: |
    grep -rnE 'print\(.*(password|token|secret)' dags/ \
      && { echo "::error::secret printed in DAG code"; exit 1; } || true

- name: No top-level Variable.get or database calls
  run: python scripts/assert_no_toplevel_io.py dags/

Cost governance

triggers:
  - type: postgresql
    metadata:
      query: >-
        SELECT ceil(count(*)::decimal / 16) FROM task_instance
        WHERE state = 'queued'
      targetQueryValue: "1"

Incident playbooks

The scheduler has stopped scheduling

Nothing is failing. Nothing is running, which is why no alert fired. Confirm with airflow jobs check --job-type SchedulerJob.

Then the four causes in order of likelihood: is the process alive; can it reach the metadata database; is one DAG file exceeding dag_file_processor_timeout and blocking a processor slot; is the database connection pool exhausted. airflow dags report names a pathological file immediately, and .airflowignore gets it out of the way while you fix it. Afterwards: heartbeat alerting, if you did not already have it.

Tasks are stuck in queued

Always capacity or routing. Check cluster parallelism, then the DAG's max_active_tasks, then pool slots with airflow pools list, then whether the tasks name a queue no running worker consumes. That last case is indistinguishable from capacity exhaustion in the UI and is not fixed by adding workers. Check airflow celery inspect active_queues against the task's queue.

The metadata database is degrading

The UI times out, the scheduler loop lengthens, everything is slow. Check size and the row counts of task_instance, log, and xcom, then the connection count against the server limit. The immediate action is db clean with a conservative cutoff. The durable fixes are scheduled retention, PgBouncer in front of Postgres when you have hundreds of Airflow connections, and moving large XCom values out. Resist resizing first. It buys weeks and hides the cause.

A backfill took production down

A secret leaked into a task log

Rotate before anything else. The log is in object storage, in a log aggregator, possibly in a Slack alert, and in browser caches. Deleting the log line is not containment. Then find the code path that printed it and close the class rather than the instance: add the key name to sensitive_var_conn_names, and add a CI grep for printing connection or variable values. Rotating without closing the class means it recurs with a different credential.

An upgrade migration is taking far longer than the window

Do not cancel it mid-migration: a partially applied schema change is worse than a long one. If you have not started, stop and reconsider: db clean first, then rehearse on a restored snapshot to get a real duration. If you are mid-migration and it is unacceptable, the fall-back is restoring the backup you took beforehand, which is why that backup must have been tested.

Running Airflow as a platform

Publish a template repository, not documentation. A working DAG with retries, timeouts, tags, and doc_md; the integrity test suite; the CI workflow with the parse-time budget. Teams inherit every decision by cloning, and a wiki page is read once.

Publish a shared operator library as a versioned package. The quality check, the alert callback, the standard extract. It is where standards get enforced: a gate that is one line gets adopted and forty lines does not.

Own the contract, not the DAGs. The platform owns scheduler health, punctuality, the shared library, and the deployment gate. DAG owners own correctness, their SLAs, and their own on-call. Writing that split down is what stops the platform team being paged for every data bug.

Measure and publish. Without numbers, "Airflow is fine" is an opinion:

Machine-learning pipelines specifically

Airflow orchestrates training; it does not train. A task that fits a model in the worker holds a scheduling slot for hours, cannot scale past one machine, and puts memory pressure on shared infrastructure. Submit to a training service, a Kubernetes job, or a managed platform, and let Airflow wait, deferrably.

Model artifacts do not belong in XCom. Pass the URI. The metadata database is not an artifact store, and a serialised model in an XCom row will slow every UI page that touches that run.

Gate promotion on metrics, in the DAG. A branch on evaluation results (promote if the new model beats the incumbent, otherwise skip and alert) turns a manual review into a recorded, auditable decision.

Datasets are the right trigger for retraining. "Retrain when the feature table is updated" expresses the real dependency; "retrain at 3am and hope features landed" does not.

Long training runs need execution_timeout more than anything else does. A hung GPU job with no timeout is the most expensive stuck task you can have.

@task.branch
def promote_or_skip(new_auc: float, incumbent_auc: float) -> str:
    return "register_model" if new_auc > incumbent_auc + 0.005 else "skip_promotion"

The checklist to run before shipping

Most orchestration incidents are prevented at review time rather than at 3am. Reading a DAG for what it guarantees, rather than what it does, is the highest-leverage habit you can build.