Top 50 Apache Airflow Interview Questions and Answers

Detailed explanation

Parsing is a loop, and it never stops:

1. Discover. Every [dag_processor] refresh_interval (300 s by default) the processor lists the files in each configured bundle. It skips paths matched by .airflowignore, which uses glob syntax by default in 3.x (regexp was the 2.x default). In safe mode it only parses files whose text contains both “airflow” and “dag”.
2. Execute. Child processes (parsing_processes, default 2) import each file. A file is re-parsed at most every min_file_process_interval (30 s). Each import is cut off after dagbag_import_timeout (30 s).
3. Serialise. Each DAG found is written to the serialized_dag table. If the structure changed, a new DAG version is recorded too (see What are DAG bundles and DAG versioning in Airflow 3?). Errors are saved as import errors and shown in the UI.
4. Schedule. The scheduler builds runs and task instances from the serialised DAG. Workers re-import the file at run time to get the real Python callables.

In Airflow 3 the DAG processor is its own required component, started with airflow dag-processor (see Airflow architecture: scheduler, executor, workers, metadata DB, API server). Because your file runs on every parse, anything at module level, such as API calls or Variable.get, runs again and again (see Why is DAG parsing slow, and how do you keep parse time low?).

Example

# Runtime: Python 3.11, Apache Airflow 3.3
# Layout: dags/orders_daily.py (valid) and dags/broken_dag.py (two tasks named "a")
import time
from airflow.dag_processing.dagbag import DagBag

start = time.perf_counter()
bag = DagBag(dag_folder="dags")   # 3.x DagBag no longer takes include_examples
print(f"parsed in {time.perf_counter() - start:.2f}s")
print("dags:", sorted(bag.dag_ids))
for path, err in bag.import_errors.items():
    print("import error:", path.split("/")[-1], "->", err.strip().splitlines()[-1])

parsed in 0.02s
dags: ['orders_daily']
import error: broken_dag.py -> airflow.sdk.exceptions.DuplicateTaskIdFound: Task id 'a' has already been added to the Dag

Real-world use case

A team merged a DAG that called a REST API at module level to build its task list. The DAG processor ran that call about every 30 seconds, all day, and the vendor started rate-limiting the account. Parse times also climbed past the timeout, and the DAG kept vanishing from the UI. They moved the call inside a task and used dynamic task mapping, and parse time fell below one second.

Common mistakes and red flags

• Thinking the scheduler imports DAG files in Airflow 3. It reads serialised DAGs; the DAG processor does the parsing.
• Thinking a file is parsed once at deploy time. Parsing is continuous.
• Hiding a broken DAG with .airflowignore instead of fixing the import error.

Follow-up questions

• Where do you see parse errors? — The UI’s import-errors banner, airflow dags list-import-errors, and the DAG processor logs.
• How do you see slow files? — Run airflow dags report or watch the dag_processing.last_duration metric, which is tagged per file.
• Can one file define many DAGs? — Yes. Every DAG object the file creates is registered (see Generating many DAGs from config vs dynamic task mapping: when do you use each?).

Detailed explanation

In Airflow 3 the authoring imports live in airflow.sdk: dag, task, DAG, TaskGroup, Asset, Variable, BaseOperator and others. Old paths such as airflow.decorators are deprecated.

What @task gives you:

• Implicit dependencies. load(transform(extract())) builds extract >> transform >> load.
• Implicit XCom. The return value is pushed as return_value. Arguments that are XComArgs are resolved before the function runs.
• Typed, testable functions. The business logic is plain Python, and .function gives you the raw callable for unit tests (see How do you test task logic and run a DAG end to end with dag.test()?).
• Variants. @task.bash, @task.branch, @task.short_circuit, @task.sensor, @task.virtualenv, @task.docker, @task.kubernetes. Each one wraps the matching operator.

How the two styles talk to each other:

Calling a decorated function does not run it. Inside the DAG body, double(3) only adds a task to the DAG. The body runs at parse time; task code runs on a worker.

Example

# Runtime: Python 3.11, Apache Airflow 3.3 (apache-airflow-providers-standard)
from datetime import datetime
from airflow.sdk import dag, task
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator


def _count_rows(**context):
    # classic style: return value is pushed to XCom as "return_value"
    return 3


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def mixed_styles():
    count = PythonOperator(task_id="count_rows", python_callable=_count_rows)

    @task
    def double(n: int) -> int:          # TaskFlow: args are pulled from XCom for you
        return n * 2

    notify = BashOperator(
        task_id="notify",
        bash_command="echo doubled={{ ti.xcom_pull(task_ids='double') }}",
    )

    doubled = double(count.output)       # .output is an XComArg for the classic task
    doubled >> notify                     # XComArg works with >> as well


dag_obj = mixed_styles()

if __name__ == "__main__":
    print(dag_obj.test().state)

... Done. Returned value was: 3
... Done. Returned value was: 6
... Running command: ['/usr/bin/bash', '-c', 'echo doubled=6']
success

Real-world use case

A team kept SQLExecuteQueryOperator from the provider for its warehouse statements. It wrote the custom business rules, such as parsing vendor files and computing reconciliation totals, as @task functions. The rule code got pytest coverage, while the integration code was the tested provider version.

Common mistakes and red flags

• Expecting result = extract() to hold real data inside the DAG body.
• Returning a large DataFrame from @task. Return values become XComs (see What are XComs, and what are their limits?, How do TaskFlow tasks pass values: return values, multiple_outputs and named XComs?).
• Wrapping PythonOperator around a provider hook when a provider operator already exists.

Follow-up questions

• How do you get the context in a TaskFlow task? — Add **context or named arguments such as ti and data_interval_start, or call get_current_context().
• How do you return several XComs? — Use @task(multiple_outputs=True) with a dict, or a TypedDict return annotation (see How do TaskFlow tasks pass values: return values, multiple_outputs and named XComs?).

Detailed explanation

Key points:

• Even “core” operators sit in providers in Airflow 3. BashOperator, PythonOperator and EmptyOperator live in apache-airflow-providers-standard, so import them from airflow.providers.standard....
• Providers are released on their own schedule. Pin them in a constraints file, so an upgrade happens on purpose and a rebuild does not change them silently.
• A custom operator subclasses BaseOperator and implements execute(context). Use template_fields to choose which attributes Jinja renders. The constructor runs at parse time, so keep it cheap and do I/O only in execute.
• Write a hook when several operators or tasks need the same client logic. Write an operator when a team repeats the same task shape. A plain @task that calls the hook is often enough.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
import json
import os
from datetime import datetime, timezone
from airflow.sdk import BaseHook, BaseOperator, dag

# A connection defined as an environment variable (JSON form); in production it
# would live in a secrets backend. The conn_id is the part after AIRFLOW_CONN_.
os.environ["AIRFLOW_CONN_INVENTORY_API"] = json.dumps(
    {"conn_type": "http", "host": "inventory.internal", "login": "svc", "password": "s3cret",
     "extra": {"page_size": 500}}
)


class InventoryHook(BaseHook):
    """Hook: knows HOW to talk to the system. Reusable by many operators."""

    def __init__(self, conn_id: str = "inventory_api"):
        super().__init__()
        self.conn_id = conn_id

    def get_client_config(self) -> dict:
        conn = self.get_connection(self.conn_id)
        return {"base_url": f"https://{conn.host}", "user": conn.login,
                "page_size": conn.extra_dejson.get("page_size", 100)}


class InventorySnapshotOperator(BaseOperator):
    """Operator: WHAT one task does, using the hook."""

    template_fields = ("snapshot_date",)

    def __init__(self, snapshot_date: str, conn_id: str = "inventory_api", **kwargs):
        super().__init__(**kwargs)
        self.snapshot_date = snapshot_date
        self.conn_id = conn_id

    def execute(self, context):
        cfg = InventoryHook(self.conn_id).get_client_config()
        print(f"GET {cfg['base_url']}/snapshot?date={self.snapshot_date} "
              f"as {cfg['user']} page_size={cfg['page_size']}")
        return {"date": self.snapshot_date, "rows": 1200}


@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False)
def inventory_snapshot():
    InventorySnapshotOperator(task_id="snapshot", snapshot_date="{{ ds }}")


dag_obj = inventory_snapshot()

if __name__ == "__main__":
    print(dag_obj.test(logical_date=datetime(2026, 3, 1, tzinfo=timezone.utc)).state)

GET https://inventory.internal/snapshot?date=2026-03-01 as svc page_size=500
success

Real-world use case

A platform team wrapped its internal data-catalogue API in a hook. It added two operators on top: register a table, and wait for an approval (built as a deferrable operator). It shipped them as an internal provider package with its own version. Dozens of teams used the same code, and a credential rotation only meant changing one connection.

Common mistakes and red flags

• Opening network connections in an operator’s __init__, which runs at every parse.
• Hard-coding credentials in the operator instead of using a conn_id.
• Saying “Airflow processes the data”. Hooks and operators usually tell another system to do the work.

Follow-up questions

• What is template_fields? — The list of attributes Jinja renders before execute() (see How does Jinja templating work in Airflow, and which macros should you know?).
• How do you find a provider’s version and hooks? — Run airflow providers list, airflow providers hooks, or check the provider’s docs and changelog.

Detailed explanation

Lookup order: the secrets backend first, if one is configured (see How do secrets backends work in Airflow 3, and what is the lookup order?), then environment variables, then the metadata database. Each environment gets its own values under the same IDs, so one DAG file works in dev, staging and prod.

Ways to read them:

In Airflow 3, task code can no longer query the metadata DB. Variable.get and get_connection in a task go through the Task Execution API (see What are the Task SDK and the Task Execution API, and why do they matter?), or through a secrets backend configured for workers.

Good practice:

• Keep secrets in a secrets backend and plain settings in Variables or code. Don’t store one huge JSON Variable that every DAG reads.
• Values of connection passwords and of Variables whose names contain words like password, secret or token are masked in task logs. Treat masking as a safety net, not a design.
• Use one conn_id naming convention across environments, for example warehouse everywhere, with different values.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
# In real deployments these come from a secrets backend or the UI; env vars keep the demo self-contained.
import os
from datetime import datetime
from airflow.sdk import Variable, dag, task
from airflow.providers.standard.operators.bash import BashOperator

os.environ["AIRFLOW_VAR_EXPORT_CONFIG"] = '{"region": "eu-west-1", "batch_size": 5000}'
os.environ["AIRFLOW_CONN_WAREHOUSE"] = "postgres://etl_user:[email protected]:5432/analytics"


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def config_demo():
    @task
    def read_config():
        # Read inside the task (at run time), never at module top level.
        cfg = Variable.get("export_config", deserialize_json=True)
        missing = Variable.get("not_set_yet", default="fallback")
        print("batch_size:", cfg["batch_size"], "| missing:", missing)

    show = BashOperator(
        task_id="templated",
        bash_command=(
            "echo region={{ var.json.export_config.region }} "
            "host={{ conn.warehouse.host }} schema={{ conn.warehouse.schema }}"
        ),
    )
    read_config() >> show


dag_obj = config_demo()

if __name__ == "__main__":
    print(dag_obj.test().state)

batch_size: 5000 | missing: fallback
... region=eu-west-1 host=wh.internal schema=analytics
success

Real-world use case

An analytics platform ran dev, staging and prod from the same Git branch. Each environment’s secrets backend held a warehouse connection pointing at its own database. A DAG promoted to prod needed no code changes, and nobody had to copy credentials into the UI.

Common mistakes and red flags

• Calling Variable.get at module level. That is a lookup on every parse, for every file.
• Putting passwords in Variables or DAG code instead of connections or a secrets backend.
• Editing connections in the UI when they actually come from a secrets backend. The UI shows only DB-stored values.

Follow-up questions

• How do you write a connection as an env var? — As a URI (postgres://user:pw@host:5432/db) or as JSON in AIRFLOW_CONN_<ID>.
• Can a task change a Variable? — Yes, Variable.set, but shared mutable state between runs is a smell. Prefer XCom, assets or the 3.3 task state store.

Detailed explanation

Rules interviewers check:

• [b, c] >> [d, e] is a TypeError, because Python cannot shift a list by a list. Use cross_downstream or chain_linear.
• Dependencies only order tasks. Whether a downstream task runs depends on its trigger rule (see What are trigger rules, and which ones do you actually use?).
• Dependencies cannot cross DAGs. For that, use assets, TriggerDagRunOperator or ExternalTaskSensor (see TriggerDagRunOperator, ExternalTaskSensor or assets: how do you choose for cross-DAG dependencies?).
• A cycle is caught at parse time and becomes an import error.
• setup and teardown tasks add a special kind of relationship (see What are setup and teardown tasks, and how are they different from an all_done cleanup?).

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import DAG, Label, chain, cross_downstream
from airflow.providers.standard.operators.empty import EmptyOperator

with DAG("deps_demo", schedule=None, start_date=datetime(2026, 1, 1), catchup=False) as dag:
    start, end = EmptyOperator(task_id="start"), EmptyOperator(task_id="end")
    extract = [EmptyOperator(task_id=f"extract_{s}") for s in ("crm", "erp")]
    clean = [EmptyOperator(task_id=f"clean_{s}") for s in ("crm", "erp")]
    checks = [EmptyOperator(task_id=f"check_{c}") for c in ("rows", "nulls")]
    publish = EmptyOperator(task_id="publish")

    # chain() with equal-length lists pairs them element-wise: extract_crm >> clean_crm, ...
    chain(start, extract, clean)
    # cross_downstream: every clean task feeds every check task (2 x 2 edges)
    cross_downstream(clean, checks)
    # >> with a list fans in; Label only annotates the edge in the UI
    checks >> Label("all checks passed") >> publish >> end

if __name__ == "__main__":
    for t in sorted(dag.tasks, key=lambda t: t.task_id):
        print(f"{t.task_id:12} -> {sorted(t.downstream_task_ids)}")

check_nulls  -> ['publish']
check_rows   -> ['publish']
clean_crm    -> ['check_nulls', 'check_rows']
clean_erp    -> ['check_nulls', 'check_rows']
end          -> []
extract_crm  -> ['clean_crm']
extract_erp  -> ['clean_erp']
publish      -> ['end']
start        -> ['extract_crm', 'extract_erp']

Real-world use case

A finance pipeline pulled data from two source systems in parallel, cleaned each one, and ran a shared set of quality checks on both. chain kept each source’s extract and clean tasks paired, and cross_downstream made every check wait for every cleaned source. Printing downstream_task_ids in a unit test then locked the structure in place.

Common mistakes and red flags

• Writing [a, b] >> [c, d] and being surprised by the error.
• Expecting chain(a, [b1, b2], [c1, c2, c3]) to work. Adjacent lists must be the same length.
• Adding edges “just in case”. Every extra edge serialises work that could run in parallel.

Follow-up questions

• How do you test the structure? — Load the DAG and assert on task.downstream_task_ids or dag.task_dict (see How do you write DAG integrity tests that catch broken DAGs before merge?).
• Do the edges between TaskFlow calls also pass data? — Yes. Passing an output both adds the edge and moves the XCom value. A bare >> only orders the tasks.

Detailed explanation

What task groups give you:

• Reuse. A @task_group function is a template. Call it once per source with .override(group_id=...) to stamp out identical pipelines.
• Group-level default_args. For example, give everything in the ingest group retries=2.
• Dependencies on the whole group. group_a >> group_b connects group_a’s leaf tasks to group_b’s root tasks.
• Readable graphs. Fifty tasks become five boxes.
• Mapping. A task group can be mapped with expand() to repeat a whole sub-pipeline per input (see How do you map a whole task group, filter inputs and limit fan-out?).

What they don’t do:

• They add no separate concurrency limit or pool. Use pools (see What are pools, and how do you use them to protect a shared system?) or max_active_tis_per_dagrun.
• They don’t isolate failures. A failure inside a group propagates like any other.
• prefix_group_id=False removes the prefix, but then task_ids must be unique across the whole DAG.

SubDAGs versus task groups:

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import dag, task, task_group


@task
def extract(source: str) -> str:
    return f"s3://raw/{source}/"

@task
def validate(path: str) -> str:
    return path

@task
def load(path: str) -> None:
    print("loading", path)


@task_group(default_args={"retries": 2})       # group-level defaults apply to its tasks
def ingest(source: str):
    load(validate(extract(source)))


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def ingest_sources():
    @task
    def publish():
        print("all sources loaded")

    groups = [ingest.override(group_id=f"ingest_{s}")(s) for s in ("crm", "billing")]
    groups >> publish()


dag_obj = ingest_sources()

if __name__ == "__main__":
    print(sorted(dag_obj.task_ids))
    print("retries:", dag_obj.get_task("ingest_crm.load").retries)
    print("publish upstream:", sorted(dag_obj.get_task("publish").upstream_task_ids))
    print(dag_obj.test().state)

['ingest_billing.extract', 'ingest_billing.load', 'ingest_billing.validate', 'ingest_crm.extract', 'ingest_crm.load', 'ingest_crm.validate', 'publish']
retries: 2
publish upstream: ['ingest_billing.load', 'ingest_crm.load']
loading s3://raw/billing/
loading s3://raw/crm/
success

Real-world use case

A marketing-data team pulled from six ad platforms, all with the same extract, validate and load shape. One @task_group function, called once per platform from a config list, replaced six copied blocks. A fix to validation now reached every platform in one change.

Common mistakes and red flags

• Expecting a task group to limit concurrency or retry as a unit.
• Using a short task id in a branch’s return value when the task lives inside a group.
• Calling task groups “SubDAGs” or proposing SubDAGs in Airflow 3.

Follow-up questions

• How do you retry a whole group? — Clear the group’s tasks in the UI. Clearing a task can also clear its downstream tasks, and setup/teardown pairs inside it (see What are setup and teardown tasks, and how are they different from an all_done cleanup?).
• Task group or separate DAG? — Use a separate DAG when the part has its own schedule, owner or SLA, and link it with assets (see How does asset-aware scheduling work, and how do you combine asset conditions?).

Detailed explanation

Mapping happens at run time. The DAG file says “one copy of export per region”, and the scheduler creates the copies after list_regions has pushed its list. The DAG’s shape stays fixed, so parsing stays fast (compare Q521).

The building blocks:

Behaviour to remember:

• An empty list creates no instances, and the mapped task is marked skipped.
• The downstream “reduce” task gets a LazySelectSequence. Iterate it or call list() on it once; don’t index into it over and over.
• [core] max_map_length (1024 by default) caps the number of instances. Cap concurrency with max_active_tis_per_dag or max_active_tis_per_dagrun.
• Each instance retries, fails and can be cleared on its own. You can rerun index 2 without touching 0 and 1.
• Only arguments that are mapped can vary per instance. Task settings such as retries, pool and queue go in partial() and apply to all instances.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import dag, task


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def mapping_basics():
    @task
    def list_regions() -> list[str]:
        return ["eu", "us", "apac"]          # decided at run time, not parse time

    # partial(): constant for every copy. expand(): one copy per element.
    @task(map_index_template="{{ task.op_kwargs['region'] }}")   # nicer UI label than 0/1/2
    def export(region: str, fmt: str) -> int:
        rows = {"eu": 120, "us": 300, "apac": 80}[region]
        print(f"export {region} as {fmt}: {rows} rows")
        return rows

    @task
    def total(counts) -> None:                # receives a lazy sequence of all results
        print("total rows:", sum(counts))

    counts = export.partial(fmt="parquet").expand(region=list_regions())
    total(counts)

    # expand over two lists = cross product; expand_kwargs = explicit pairs
    @task
    def copy(src: str, dst: str):
        print(f"copy {src} -> {dst}")

    copy.expand_kwargs([{"src": "a.csv", "dst": "bronze/a"}, {"src": "b.csv", "dst": "bronze/b"}])


dag_obj = mapping_basics()

if __name__ == "__main__":
    run = dag_obj.test()
    print(sorted((ti.task_id, ti.map_index, ti.rendered_map_index) for ti in run.get_task_instances()
                 if ti.map_index >= 0))
    print(run.state)

copy a.csv -> bronze/a
copy b.csv -> bronze/b
export us as parquet: 300 rows
export apac as parquet: 80 rows
export eu as parquet: 120 rows
total rows: 500
[('copy', 0, '0'), ('copy', 1, '1'), ('export', 0, 'eu'), ('export', 1, 'us'), ('export', 2, 'apac')]
success

Real-world use case

A logistics company got between 3 and 40 carrier files a day. One mapped load_file task replaced a nightly job that had looped over the files inside a single task. A bad file now failed only its own map index. The on-call engineer fixed the file and cleared that one instance, and the other loads kept their results. (For choosing between mapping and other designs for this case, see the site’s guide on processing a variable number of files.)

Common mistakes and red flags

• Passing two lists to expand() and expecting them to be zipped. You get a cross product.
• Listing files at parse time to build tasks, which runs on every parse.
• Mapping over 50,000 tiny items. Batch them into chunks first, because every instance carries scheduling overhead.

Follow-up questions

• How do you pull one instance’s XCom? — ti.xcom_pull(task_ids="export", map_indexes=1).
• Can classic operators be mapped? — Yes, for example BashOperator.partial(task_id="x").expand(bash_command=[...]).

In depth: How would you process a variable number of files each day in Airflow?

Detailed explanation

• What is templated: only the fields named in template_fields, such as bash_command or sql. A string ending in a template_ext extension like .sql is loaded as a template file, searched under the DAG folder and template_searchpath.
• When: at task run time on the worker. The DAG file only holds the template string, so {{ ds }} costs nothing at parse time.
• TaskFlow: name context keys as function arguments (ds=None, data_interval_start=None), or call get_current_context(). You don’t need Jinja inside Python.

Context and macros used most:

Airflow 3 removed the old keys execution_date, next_ds, prev_ds, yesterday_ds and tomorrow_ds. Use logical_date and the data_interval_* keys, plus macros.ds_add when you need date arithmetic. Templates render to strings by default. Set render_template_as_native_obj=True on the DAG, or on a single operator since 3.2, to get Python lists and dicts back.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime, timezone
from airflow.sdk import dag, task
from airflow.providers.standard.operators.bash import BashOperator


def fiscal_quarter(d) -> str:              # a custom macro: plain Python
    return f"FY{d.year}Q{(d.month - 1) // 3 + 1}"


@dag(
    schedule="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    user_defined_macros={"fiscal_quarter": fiscal_quarter},
)
def templating_demo():
    BashOperator(
        task_id="export",
        # Rendered on the worker just before execute(); the DAG file only holds the string.
        bash_command=(
            "echo ds={{ ds }} "
            "start={{ data_interval_start | ds }} "
            "week_ago={{ macros.ds_add(ds, -7) }} "
            "compact={{ ds_nodash }} "
            "fq={{ fiscal_quarter(data_interval_start) }} "
            "run={{ run_id }}"
        ),
    )

    @task
    def show_context(ds=None, data_interval_start=None, data_interval_end=None):
        # TaskFlow: name the context keys you need as arguments
        print("ds:", ds, "| interval:", data_interval_start, "->", data_interval_end)

    @task
    def removed_keys(**context):
        print("has yesterday_ds:", "yesterday_ds" in context, "| has execution_date:", "execution_date" in context)

    show_context() >> removed_keys()


dag_obj = templating_demo()

if __name__ == "__main__":
    print(dag_obj.test(logical_date=datetime(2026, 5, 20, 6, tzinfo=timezone.utc)).state)

ds: 2026-05-20 | interval: 2026-05-20 06:00:00+00:00 -> 2026-05-20 06:00:00+00:00
... ds=2026-05-20 start=2026-05-20 week_ago=2026-05-13 compact=20260520 fq=FY2026Q2 run=manual__2026-10-17T13:42:04.945864+00:00
has yesterday_ds: False | has execution_date: False
success

The interval has zero length because a cron string in Airflow 3 uses CronTriggerTimetable by default (see CronTriggerTimetable vs CronDataIntervalTimetable: what changed in Airflow 3, and when do you write a custom timetable?).

Real-world use case

A warehouse load used sql="sql/load_orders.sql", and the SQL file filtered on '{{ data_interval_start }}' <= order_ts AND order_ts < '{{ data_interval_end }}'. Clearing any past run reloaded exactly that window. The team reviewed the rendered SQL in the UI’s “Rendered Template” tab before running backfills.

Common mistakes and red flags

• Using {{ ds }} in a field that isn’t in template_fields, which leaves the literal braces in the output.
• Building dates with datetime.now() in Python instead of using the run’s interval.
• Using removed 2.x keys such as execution_date or yesterday_ds in Airflow 3.

Follow-up questions

• How do you template a field on a custom operator? — Add its attribute name to template_fields (see Operators, hooks and providers: how do they fit together?).
• Where do you see the rendered values? — The task instance’s Rendered Template view, or airflow tasks render.

Detailed explanation

Other settings that shape scheduling:

• start_date is the earliest date that can be scheduled. Use a fixed, timezone-aware date, never datetime.now(). Its timezone, or the default timezone (UTC), sets the cron’s timezone, so "0 6 * * *" with a Europe/London start date follows daylight-saving changes.
• catchup is False by default in Airflow 3 ([scheduler] catchup_by_default). With True, Airflow creates every missed run since start_date (see Airflow catchup, backfill, start_date and logical date).
• end_date stops scheduling after a date.
• max_active_runs limits concurrent runs of the DAG (see How do Airflow’s concurrency limits stack, from parallelism down to a single task?).

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime, timedelta
from airflow.sdk import DAG, Asset

schedules = {
    "None (manual/API only)": None,
    '"@daily" preset': "@daily",
    '"15 2 * * 1-5" cron': "15 2 * * 1-5",
    "timedelta(hours=4)": timedelta(hours=4),
    '"@once"': "@once",
    '"@continuous"': "@continuous",
    "[Asset(...)]": [Asset("s3://lake/orders/")],
}
for label, sched in schedules.items():
    kwargs = {"max_active_runs": 1} if sched == "@continuous" else {}
    d = DAG(f"d{abs(hash(label))}", schedule=sched, start_date=datetime(2026, 1, 1), **kwargs)
    print(f"{label:24} -> {type(d.timetable).__name__}")

None (manual/API only)   -> NullTimetable
"@daily" preset          -> CronTriggerTimetable
"15 2 * * 1-5" cron      -> CronTriggerTimetable
timedelta(hours=4)       -> DeltaTriggerTimetable
"@once"                  -> OnceTimetable
"@continuous"            -> ContinuousTimetable
[Asset(...)]             -> AssetTriggeredTimetable

Real-world use case

A retail team scheduled store-sales loads at "30 5 * * *" with a start_date in America/New_York. The loads stayed at 05:30 local time through the daylight-saving changes, so the morning reports were ready when stores opened, without anyone editing UTC offsets twice a year.

Common mistakes and red flags

• Using start_date=datetime.now(). The start date moves on every parse and runs never begin.
• Using schedule_interval or timetable=, which were removed in Airflow 3. Use schedule.
• Expecting "@daily" in Airflow 3 to run at midnight for “yesterday” with a one-day data interval. That is the old CronDataIntervalTimetable behaviour (see CronTriggerTimetable vs CronDataIntervalTimetable: what changed in Airflow 3, and when do you write a custom timetable?).

Follow-up questions

• How do you see the next run? — airflow dags next-execution <dag_id>, or the DAG header in the UI.
• How do you run on the 3rd business day of the month? — A custom timetable, or a daily cron plus a short-circuit (see Short-circuit, branch or AirflowSkipException: how do you skip work cleanly?).

Detailed explanation

ExternalTaskSensor looks for the same logical date in the other DAG unless you shift it. A 03:00 DAG waiting on a 02:00 DAG needs execution_delta=timedelta(hours=1). If you get this wrong, the sensor waits until it times out. Run it with mode="reschedule" or deferrable=True, and always set timeout (see Which sensor settings matter, and how do you write a sensor with @task.sensor?). In Airflow 3, logical dates can be None for manual and asset-triggered runs (see Manual and API-triggered runs in Airflow 3: why can logical_date be None?), which makes date-matching sensors even more fragile. That is one more reason to prefer assets.

In Airflow 3, these operators live in apache-airflow-providers-standard. A run created by TriggerDagRunOperator shows up with its own run type, so you can tell it apart from manual runs.

For a longer walk-through, see the site’s guide on making one Airflow DAG depend on another.

Example

# Runtime: Python 3.11, Apache Airflow 3.3 (apache-airflow-providers-standard)
from datetime import datetime, timedelta
from airflow.sdk import dag, task
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.providers.standard.sensors.external_task import ExternalTaskSensor


@dag(schedule="0 2 * * *", start_date=datetime(2026, 1, 1), catchup=False)
def load_sales():
    @task
    def load() -> int:
        return 1834

    # Push: the producer names its downstream and hands over parameters.
    load() >> TriggerDagRunOperator(
        task_id="trigger_finance_report",
        trigger_dag_id="finance_report",
        conf={"source": "load_sales", "rows": "{{ ti.xcom_pull(task_ids='load') }}"},
        wait_for_completion=False,
    )


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def finance_report():
    @task
    def build(params=None, dag_run=None):
        print("triggered with conf:", dag_run.conf)

    build()


# Pull: a DAG on its own schedule waits for a task in another DAG.
@dag(schedule="0 3 * * *", start_date=datetime(2026, 1, 1), catchup=False)
def marketing_mart():
    ExternalTaskSensor(
        task_id="wait_for_sales_load",
        external_dag_id="load_sales",
        external_task_id="load",
        execution_delta=timedelta(hours=1),     # 03:00 run looks for the 02:00 run
        mode="reschedule", poke_interval=300, timeout=3 * 3600,
    )


producer, consumer, sensor_dag = load_sales(), finance_report(), marketing_mart()

if __name__ == "__main__":
    print("producer:", producer.test().state)
    from airflow.models.dagrun import DagRun
    for dr in DagRun.find(dag_id="finance_report"):
        print("finance_report run:", dr.run_type, dr.state, dr.conf)

producer: success
finance_report run: operator_triggered queued {'source': 'load_sales', 'rows': '1834'}

The triggered run waits in queued because no scheduler is running in this local test. rows arrives as a string because templates render to strings (see How does Jinja templating work in Airflow, and which macros should you know?).

Real-world use case

A company had twelve report DAGs that each used an ExternalTaskSensor on one ingestion DAG, so every schedule change broke some sensors. They replaced the sensors with an asset on the ingestion output. The twelve reports now run when the data lands, and the ingestion team can change its schedule without telling anyone.

Common mistakes and red flags

• An ExternalTaskSensor between DAGs on different schedules with no execution_delta.
• Chains of TriggerDagRunOperator(wait_for_completion=True) without deferrable mode, which tie up worker slots.
• Building a web of cross-DAG links that should have been one DAG or task groups.

Follow-up questions

• How do you see dependencies between DAGs? — The UI’s DAG dependencies view, or airflow dags show-dependencies.
• Can TriggerDagRunOperator reuse a run id? — Yes, with trigger_run_id. reset_dag_run=True clears an existing run instead of failing.

In depth: How do you make one Airflow DAG depend on another?

Detailed explanation

Things to remember:

• multiple_outputs is inferred from a dict or TypedDict return annotation. Pass multiple_outputs=False to stop it.
• Values must be serialisable. Primitives, lists, dicts, dataclasses and a few library types work through Airflow’s serde. Pickling is not the default.
• XComs belong to one DAG run and one try. Clearing a task deletes its XComs, so downstream tasks must not expect old values.
• Return pointers (paths, table names, counts, ids), not data (see What are XComs, and what are their limits?, How do custom and object-storage XCom backends work?).
• do_xcom_push=False on classic operators stops them from pushing anything, for example a huge stdout from a Bash command.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import dag, task


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def passing_data():
    @task(multiple_outputs=True)       # also inferred from a dict / TypedDict return annotation
    def extract() -> dict:
        # Return a pointer to the data, not the data itself
        return {"path": "s3://lake/raw/orders/2026-05-20.parquet", "rows": 1834}

    @task
    def validate(path: str, rows: int) -> str:
        if rows == 0:
            raise ValueError("empty extract")
        return path

    @task
    def publish(path: str, ti=None):
        ti.xcom_push(key="published_path", value=path)     # an extra, named XCom
        print("published", path)

    @task
    def audit(ti=None):
        # Pull by task id and key; TaskFlow wiring would be nicer, this shows the raw API
        print("audit saw:", ti.xcom_pull(task_ids="publish", key="published_path"),
              "| rows key:", ti.xcom_pull(task_ids="extract", key="rows"))

    meta = extract()
    publish(validate(meta["path"], meta["rows"])) >> audit()   # meta["path"] -> XComArg for one key


dag_obj = passing_data()

if __name__ == "__main__":
    print(dag_obj.test().state)

published s3://lake/raw/orders/2026-05-20.parquet
audit saw: s3://lake/raw/orders/2026-05-20.parquet | rows key: 1834
success

Real-world use case

An ingestion task returned {"path": ..., "rows": ..., "max_ts": ...}. A quality task used rows, the load task used path, and the watermark task used max_ts. Each took only the key it needed, so the graph showed exactly which value each step depended on.

Common mistakes and red flags

• xcom_pull with no task_ids, or a misspelt key, which returns None without an error.
• Expecting XComs to survive a clear, or to be shared between DAG runs.
• Returning a DataFrame “because it worked in dev”.

Follow-up questions

• Why not just use globals? — Every task runs in its own process, possibly on another machine. Only XCom, storage or the state store crosses that boundary.
• What is the 3.3 task state store for? — Key/value state that persists across retries and runs, such as a watermark. XComs, by contrast, belong to one run.

Detailed explanation

• Default. Tasks without a pool use default_pool, which has 128 slots ([core] default_pool_task_slot_count).
• Create pools with airflow pools set NAME SLOTS "description", the REST API, the UI, or your infrastructure code. Pools are deployment state, not DAG code, so manage them like config.
• Weighting. pool_slots=3 lets one heavy query count as three light ones.
• Deferred tasks don’t use pool slots unless the pool has include_deferred=True.
• Order in a full pool. Higher priority_weight goes first (see How do priority weights and weight_rule decide what runs first?).
• Mapping. Every mapped instance takes slots, so a pool is the natural limit for a large fan-out against one system (see How do you map a whole task group, filter inputs and limit fan-out?).

Pools compared with other limits:

See How do Airflow’s concurrency limits stack, from parallelism down to a single task? for how they combine.

A task that names a pool which doesn’t exist won’t be scheduled. Create pools before deploying the DAGs that use them, and test that every referenced pool exists in CI (see What does a good CI/CD pipeline for Airflow DAGs look like?).

Example

# Runtime: Apache Airflow 3.3 CLI
airflow pools set warehouse_heavy 4 "Big warehouse queries" --include-deferred
airflow pools list

Pool warehouse_heavy created
pool            | slots | description           | include_deferred | team_name
================+=======+=======================+==================+==========
default_pool    | 128   | Default pool          | False            | None
warehouse_heavy | 4     | Big warehouse queries | True             | None

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import dag, task


@dag(schedule="0 4 * * *", start_date=datetime(2026, 1, 1), catchup=False)
def nightly_marts():
    # Every mapped copy competes for the same 4 slots, across ALL DAGs using this pool.
    @task(pool="warehouse_heavy", pool_slots=1)
    def build_mart(name: str):
        print("building", name)

    # A very heavy statement can take several slots of the same pool.
    @task(pool="warehouse_heavy", pool_slots=3)
    def rebuild_fact_sales():
        print("rebuilding fact_sales")

    build_mart.expand(name=["finance", "marketing", "ops"]) >> rebuild_fact_sales()


dag_obj = nightly_marts()

if __name__ == "__main__":
    run = dag_obj.test()
    print({(ti.task_id, ti.map_index): (ti.pool, ti.pool_slots) for ti in run.get_task_instances()})
    print(run.state)

{('build_mart', 0): ('warehouse_heavy', 1), ('build_mart', 1): ('warehouse_heavy', 1), ('build_mart', 2): ('warehouse_heavy', 1), ('rebuild_fact_sales', -1): ('warehouse_heavy', 3)}
success

Real-world use case

Twenty teams’ DAGs all ran heavy SQL against one shared warehouse around 04:00, and queued queries regularly hit the warehouse’s concurrency limit. A warehouse_heavy pool with 6 slots made Airflow queue the excess itself. Teams could see their tasks waiting for a slot in the UI, instead of getting cryptic warehouse timeouts.

Common mistakes and red flags

• Using pools as a per-DAG throttle when max_active_tasks or max_active_runs is meant for that.
• Long poke-mode sensors in a small pool, which starve the real work.
• Referencing a pool in code that was never created in production.

Follow-up questions

• How do you see pool pressure? — The Pools page shows running, queued and scheduled slots. The pool.open_slots and pool.starving_tasks metrics, tagged by pool, expose it too.
• Can pools be team-scoped? — In 3.2+ multi-team deployments, pools can be assigned to a team.

Detailed explanation

Behaviour to know:

• Callbacks fire only for state changes caused by running tasks. Marking a task failed or successful in the UI or CLI doesn’t trigger them.
• In 3.0+ on_success_callback no longer fires for tasks that end skipped.
• Put task callbacks in default_args to cover every task. Put a DAG-level on_failure_callback on the DAG for one alert per run instead of one per task.
• Notifiers are reusable classes with template_fields and notify(context). Prefer provider notifiers, such as SlackNotifier or SmtpNotifier, over hand-written HTTP calls.
• A callback must be quick and must not fail. An exception in it is only logged, so the alert is lost.
• For “the run didn’t finish by a certain time”, use Deadline Alerts (see SLAs were removed in Airflow 3. How do Deadline Alerts replace them?). For alerting patterns overall, see the site’s guide on failure handling and alerting.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime, timedelta
from airflow.sdk import BaseNotifier, dag, task


class ChatNotifier(BaseNotifier):
    """Reusable alert: template_fields are rendered with the task's context, like an operator."""
    template_fields = ("message",)

    def __init__(self, message: str, channel: str = "#data-alerts"):
        super().__init__()
        self.message, self.channel = message, channel

    def notify(self, context):
        # real version: post to Slack/Teams via a provider hook or notifier
        print(f"to {self.channel}: {self.message}")


def log_retry(context):                      # plain-function callbacks work too
    ti = context["ti"]
    print(f"retrying {ti.task_id}, try {ti.try_number}")


@dag(
    schedule=None, start_date=datetime(2026, 1, 1), catchup=False,
    on_failure_callback=ChatNotifier("DAG {{ dag.dag_id }} run {{ run_id }} FAILED"),
    default_args={
        "retries": 1, "retry_delay": timedelta(seconds=1),
        "on_retry_callback": log_retry,
        "on_failure_callback": ChatNotifier("task {{ ti.task_id }} failed after {{ ti.try_number }} tries"),
    },
)
def alerting_demo():
    @task
    def load():
        raise RuntimeError("warehouse unavailable")

    load()


dag_obj = alerting_demo()

if __name__ == "__main__":
    print(dag_obj.test().state)

retrying load, try 1
to #data-alerts: task load failed after 2 tries
failed

In this local dag.test() run only the task-level callbacks printed. In a deployment, the DAG-level on_failure_callback is run by the DAG processor once the run fails.

Real-world use case

A team with 300 tasks had a task-level on_failure_callback posting to Slack. A single warehouse outage produced 300 messages, and the channel got muted. They moved the alert to DAG level with a notifier that lists the failed task ids, and kept task-level alerts only for a few critical tasks. One incident became one message.

Common mistakes and red flags

• Expecting callbacks to fire after marking a task failed in the UI.
• Doing heavy work, such as queries or retries, inside a callback.
• Hard-coding webhook URLs in DAG code instead of keeping them in a connection.

Follow-up questions

• Where do you find callback errors? — Task callbacks in the task log. DAG callbacks in the DAG processor logs.
• Can you attach several notifiers? — Yes. Pass a list, for example Slack and PagerDuty.

In depth: How do you handle task failures and alerting in Airflow?

Detailed explanation

What a DagBag load catches, because it runs the same import the DAG processor runs:

• Python errors and missing packages. CI must use the same image or constraints as production.
• DuplicateTaskIdFound, cycles, unknown executors (see How do multiple executors work, and how do you route a task to a specific one?), invalid schedule values and failed param defaults.
• Removed Airflow 2 APIs, such as airflow.datasets or schedule_interval (see Walk through migrating a real Airflow 2 deployment to Airflow 3. What breaks first?).

Convention checks worth adding:

Practical notes:

• In 3.3, DagBag(dag_folder=...) no longer takes include_examples. Example DAGs are separate bundles.
• Keep test files outside the DAG folder. A file that mentions “airflow” and “dag” inside the folder gets parsed as a DAG file.
• Use dag.test() for behaviour and plain unit tests for logic (see How do you test task logic and run a DAG end to end with dag.test()?). Integrity tests only prove that a DAG loads.

Example

# Runtime: Python 3.11, Apache Airflow 3.3, pytest — tests/test_dag_integrity.py, run in CI on every PR
import os
import time
import pytest
from airflow.dag_processing.dagbag import DagBag

DAG_FOLDER = os.path.join(os.path.dirname(__file__), "..", "dags")


@pytest.fixture(scope="session")
def dagbag():
    start = time.perf_counter()
    bag = DagBag(dag_folder=DAG_FOLDER)
    bag.parse_seconds = time.perf_counter() - start
    return bag


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


def test_parse_time_budget(dagbag):
    assert dagbag.parse_seconds < 5, f"DAG folder took {dagbag.parse_seconds:.1f}s to parse"


@pytest.mark.parametrize("dag_id", sorted(DagBag(dag_folder=DAG_FOLDER).dag_ids))
def test_dag_conventions(dagbag, dag_id):
    dag = dagbag.get_dag(dag_id)
    assert any(t.startswith("team:") for t in dag.tags), f"{dag_id}: add a team:<name> tag"
    for t in dag.tasks:
        assert t.retries >= 1, f"{dag_id}.{t.task_id}: set retries (default_args)"

# dags/ holds orders_daily (tagged, retries=2) and adhoc_export (no tags, no retries)
>       assert any(t.startswith("team:") for t in dag.tags), f"{dag_id}: add a team:<name> tag"
E       AssertionError: adhoc_export: add a team:<name> tag
...
FAILED tests/test_dag_integrity.py::test_dag_conventions[adhoc_export] - Asse...
1 failed, 3 passed in 1.78s

Real-world use case

A platform team saw roughly one “DAG disappeared” incident a week, caused by typos and missing provider packages in merged code. A required CI check that runs these tests in the production image made such merges impossible, and the incidents stopped. The convention tests also made ownership tags universal within a month.

Common mistakes and red flags

• Running CI in an environment that doesn’t match production, so it passes in CI and breaks in prod.
• Treating “it loads” as “it works”. Pair integrity tests with dag.test() and unit tests.
• Calling real APIs or databases at import time, which makes CI depend on the network.

Follow-up questions

• How do you test only the changed DAGs? — Load a DagBag per changed file (DagBag(dag_folder=path)), but keep a nightly full-folder check.
• Where do linters fit? — ruff with the AIR rules catches deprecated or removed Airflow APIs before the tests run (see What does a good CI/CD pipeline for Airflow DAGs look like?, Walk through migrating a real Airflow 2 deployment to Airflow 3. What breaks first?).

In depth: How do you test Airflow DAGs before they reach production?

Detailed explanation

DAG runs have only queued, running, success and failed. A run’s final state comes from its leaf tasks (teardowns excepted).

Tools for each state: the task-instance details panel in the UI, which lists unmet dependencies; airflow tasks failed-deps; airflow tasks states-for-dag-run; and the scheduler and executor logs for queued problems. For a full triage of slow or stuck tasks, see the site’s guide.

Example

# Runtime: Apache Airflow 3.3 CLI — inspect the run from the trigger-rule example (see Q514)
airflow dags list-runs trigger_rules_demo -o plain
airflow tasks states-for-dag-run trigger_rules_demo "manual__2026-10-17T13:37:18.923999+00:00"

dag_id              run_id                                    state    run_after ...
trigger_rules_demo  manual__2026-10-17T13:37:18.923999+00:00  failed   2026-10-17T13:37:18.923999+00:00 ...
dag_id             | logical_date                     | task_id      | state           | start_date ...
===================+==================================+==============+=================+===========
trigger_rules_demo | 2026-10-17T13:37:18.718414+00:00 | load_eu      | success         | ...
trigger_rules_demo | 2026-10-17T13:37:18.718414+00:00 | load_us      | failed          | ...
trigger_rules_demo | 2026-10-17T13:37:18.718414+00:00 | build_report | upstream_failed | ...
trigger_rules_demo | 2026-10-17T13:37:18.718414+00:00 | cleanup_tmp  | success         | ...
trigger_rules_demo | 2026-10-17T13:37:18.718414+00:00 | page_on_call | success         | ...

Real-world use case

Every morning dozens of tasks sat in queued for 20 minutes. Their state showed the scheduler had done its part, so the problem was after the hand-off. The Kubernetes events showed pods Pending because a node pool had hit its autoscaling maximum. Raising the limit fixed it, with no Airflow configuration change.

Common mistakes and red flags

• Treating scheduled and queued as the same thing. They point to different bottlenecks.
• Clearing tasks over and over without reading the unmet dependencies.
• Not knowing deferred needs a running triggerer.

Follow-up questions

• What does removed mean after a deploy? — The task_id no longer exists in the DAG’s version. With versioned bundles, old runs still show the old graph.
• How do you force a stuck task to run again? — Find the cause first, then clear it. Marking it success or failed by hand skips callbacks.

Detailed explanation

The rules in Airflow 3.3’s TriggerRule enum:

Two behaviours interviewers probe:

• Propagation. Under all_success, a failure upstream marks the child upstream_failed, and a skip marks it skipped. Each state cascades down until some task’s rule accepts it.
• DAG run state comes from the leaf tasks. If the only leaf is an all_done cleanup task that succeeds, the run can be marked success even though real work failed. Either keep a work task as a leaf or make the cleanup a teardown, because teardowns are ignored when the run’s state is worked out.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import TriggerRule, dag, task


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def trigger_rules_demo():
    @task
    def load_eu():
        return "ok"

    @task
    def load_us():
        raise ValueError("source file missing")

    @task  # default all_success: will not run because load_us failed
    def build_report():
        print("report built")

    @task(trigger_rule=TriggerRule.ALL_DONE)  # runs whatever happened upstream
    def cleanup_tmp():
        print("temp files removed")

    @task(trigger_rule=TriggerRule.ONE_FAILED)  # runs as soon as any upstream fails
    def page_on_call():
        print("paging on-call")

    loads = [load_eu(), load_us()]
    loads >> build_report()
    loads >> cleanup_tmp()
    loads >> page_on_call()


dag_obj = trigger_rules_demo()

if __name__ == "__main__":
    run = dag_obj.test()
    for ti in sorted(run.get_task_instances(), key=lambda t: t.task_id):
        print(f"{ti.task_id:14} {ti.state}")
    print("dag run:", run.state)

temp files removed
paging on-call
build_report   upstream_failed
cleanup_tmp    success
load_eu        success
load_us        failed
page_on_call   success
dag run: failed

Real-world use case

A nightly job ended in a single all_done task that dropped staging tables. For weeks the runs showed green while the load step was failing, because that cleanup task was the only leaf and it succeeded. The team turned the cleanup into a teardown and added a one_failed alert task. After that, the failures showed up the same night.

Common mistakes and red flags

• Leaving the join after a branch on the default rule, so it is always skipped.
• Using always or all_done on tasks that need correct upstream data.
• Assuming a trigger rule can look at tasks that are not direct parents.

Follow-up questions

• What is the difference between none_failed and none_failed_min_one_success? — The second also requires at least one success, so it won’t run when every upstream was skipped.
• How do you alert on failure without a task? — Use on_failure_callback or a notifier (see How do callbacks and notifiers work, and where do they run?).

Detailed explanation

How it works:

1. The branch callable runs and returns "a", ["a", "b"], or None/[]. Returning nothing skips every direct downstream task.
2. Every task directly downstream of the branch that was not chosen is set to skipped.
3. Each later task applies its own trigger rule to its parents’ states (see What are trigger rules, and which ones do you actually use?).

Variants:

Design rules:

• Branch on run metadata, such as data_interval_start, params or upstream XComs, not on datetime.now(). A rerun next week must take the same path.
• Return task_ids that are direct downstream tasks. Inside a task group, use the full id, such as "group.task".
• Keep the branch cheap. A heavy check belongs in an upstream task whose result the branch reads.
• If the branches build the same output in different ways, think about one task with a parameter instead. Branches add a lot of skipped boxes to the grid.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime, timezone
from airflow.sdk import TriggerRule, dag, task


@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False)
def branch_by_day():
    @task.branch
    def choose_load(**context) -> str:
        # Branch on run metadata, not on wall-clock time, so reruns pick the same path.
        is_month_start = context["data_interval_start"].day == 1
        return "full_reload" if is_month_start else "incremental_load"

    @task
    def full_reload():
        print("rebuilding whole table")

    @task
    def incremental_load():
        print("merging one day")

    # Without this trigger rule the join would be skipped along with the unchosen branch.
    @task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
    def refresh_dashboard():
        print("dashboard refreshed")

    choose_load() >> [full_reload(), incremental_load()] >> refresh_dashboard()


dag_obj = branch_by_day()

if __name__ == "__main__":
    for day in (1, 2):
        run = dag_obj.test(logical_date=datetime(2026, 3, day, tzinfo=timezone.utc))
        states = {ti.task_id: ti.state for ti in run.get_task_instances()}
        print(f"2026-03-0{day}:", dict(sorted(states.items())))

2026-03-01: {'choose_load': 'success', 'full_reload': 'success', 'incremental_load': 'skipped', 'refresh_dashboard': 'success'}
2026-03-02: {'choose_load': 'success', 'full_reload': 'skipped', 'incremental_load': 'success', 'refresh_dashboard': 'success'}

Real-world use case

A subscription-metrics DAG fully rebuilt its churn table on the first day of each month and merged one day on the other days. Because the branch read data_interval_start, a backfill of last quarter took the full-reload path on exactly the right three dates.

Common mistakes and red flags

• Forgetting the join’s trigger rule, then “fixing” it with all_done, which also runs after a failure.
• Branching on datetime.now() or on a Variable someone edits, which makes reruns take a different path.
• Returning a task_id that is not a direct child of the branch task.

Follow-up questions

• Can a branch pick several paths? — Yes. Return a list of task_ids.
• Branch or dynamic task mapping? — Branch picks from a fixed set of paths. Mapping runs one task over a variable number of inputs (see How do partial() and expand() work in dynamic task mapping, and what is a map index?).

Detailed explanation

Points to make:

• A skip is not a failure. A DAG run whose leaves are all success or skipped counts as successful. Make sure “no data” really is fine; sometimes it should be an alert instead.
• A short-circuit that runs before a sensor saves the waiting time as well as the work.
• on_skipped_callback fires only when a task raises AirflowSkipException while running. It does not fire for tasks skipped by a branch or a trigger rule.
• In Airflow 3, import task exceptions from airflow.sdk.exceptions.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import TriggerRule, dag, task
from airflow.sdk.exceptions import AirflowSkipException


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def short_circuit_demo():
    @task
    def count_new_rows() -> int:
        return 0                      # nothing arrived today

    # False/falsy return -> skip downstream. ignore_downstream_trigger_rules=False
    # skips only direct downstream tasks and lets the others apply their own trigger rules.
    @task.short_circuit(ignore_downstream_trigger_rules=False)
    def has_new_rows(n: int) -> bool:
        return n > 0

    @task
    def transform():
        print("transforming")

    @task(trigger_rule=TriggerRule.ALL_DONE)
    def write_audit_row():
        print("audit row written")       # still runs: it has its own trigger rule

    @task
    def optional_export():
        if True:                          # e.g. feature flag off for this tenant
            raise AirflowSkipException("export disabled for this tenant")

    gate = has_new_rows(count_new_rows())
    gate >> transform() >> write_audit_row()
    optional_export()


dag_obj = short_circuit_demo()

if __name__ == "__main__":
    run = dag_obj.test()
    for ti in sorted(run.get_task_instances(), key=lambda t: t.task_id):
        print(f"{ti.task_id:16} {ti.state}")
    print("dag run:", run.state)

audit row written
count_new_rows   success
has_new_rows     success
optional_export  skipped
transform        skipped
write_audit_row  success
dag run: success

With the default ignore_downstream_trigger_rules=True, the same run also marks write_audit_row as skipped.

Real-world use case

A partner-feed DAG ran every hour, but the partner only published a few times a day. A short-circuit on “new manifest present” cut the run to two cheap tasks on most hours. The ignore_downstream_trigger_rules=False setting kept the audit-log task running, so operators could see each empty hour in the audit table.

Common mistakes and red flags

• Expecting a downstream all_done task to run after a default short-circuit.
• Using a skip to hide errors. If the data should have arrived, fail or alert.
• Catching every exception and turning it into AirflowSkipException.

Follow-up questions

• How do you skip one mapped instance? — Raise AirflowSkipException inside it. Only that map index is skipped.
• What does AirflowFailException do differently? — It fails the task without using the remaining retries.

Detailed explanation

Mark tasks with @setup / @teardown, or call .as_setup() / .as_teardown(setups=...) on existing operators. The tasks between a pair are in its scope.

Why this beats a plain cleanup task with trigger_rule="all_done":

Rules:

• A teardown’s trigger rule cannot be changed.
• Every task in a setup’s scope implicitly needs that setup to succeed.
• Setups and teardowns can be attached to a whole task group, which gives the group a clear lifecycle.
• They are tasks, so they use worker slots and honour retries. Keep teardowns idempotent (“delete if exists”), because they may run twice.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import dag, setup, task, teardown


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def ephemeral_cluster():
    @setup
    def create_cluster() -> str:
        print("cluster created")
        return "cluster-42"

    @task
    def run_job(cluster_id: str):
        raise RuntimeError(f"job failed on {cluster_id}")

    @task
    def publish():
        print("published")

    @teardown
    def delete_cluster(cluster_id: str):
        print(f"deleted {cluster_id}")       # runs even though run_job failed

    cluster = create_cluster()
    job = run_job(cluster)
    job >> publish()
    # as_teardown links the setup to the teardown; the work in between is "in scope"
    job >> delete_cluster(cluster).as_teardown(setups=cluster)


dag_obj = ephemeral_cluster()

if __name__ == "__main__":
    run = dag_obj.test()
    for ti in sorted(run.get_task_instances(), key=lambda t: t.task_id):
        print(f"{ti.task_id:15} {ti.state}")
    print("dag run:", run.state)

cluster created
deleted cluster-42
create_cluster  success
delete_cluster  success
publish         upstream_failed
run_job         failed
dag run: failed

Real-world use case

A team spun up a short-lived Spark cluster for each nightly run. Before setup/teardown, a failed job left the cluster running until someone noticed it on the bill. After the change, the delete always ran. Clearing the failed job also created a new cluster, instead of trying to reuse the one that had been deleted.

Common mistakes and red flags

• Using all_done for cleanup and then wondering why failed runs show green.
• Writing a teardown that fails if the resource is already gone.
• Expecting teardown failures to fail the run without on_failure_fail_dagrun=True.

Follow-up questions

• What happens if the setup fails? — The work tasks become upstream_failed, and the teardown does not run, because no setup succeeded.
• Can one teardown serve several setups? — Yes. Pass a list to as_teardown(setups=[...]).

Detailed explanation

Rules for DAG factories:

• The config travels with the code. Use a YAML file in the bundle, not a database query or an API call. Parse-time I/O runs every 30 seconds or so, and if it fails, DAGs disappear.
• Use a plain builder function that takes Python values and returns a DAG. Arguments of an @dag-decorated function become DAG params, not Python values. Passing a list that way and then calling .expand() on it fails at parse time with expand() got an unexpected type 'DagParam'.
• Keep dag_ids stable. Renaming a generated id orphans its history.
• get_parsing_context().dag_id is set when a worker parses the file to run one task. Skipping the other DAGs then speeds up task start. The DAG processor leaves it as None and builds all of them.
• Libraries such as DAG Factory build DAGs from YAML. They are useful for platform teams, but they still parse as Python.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
# One file -> one DAG per tenant. The config is a static file shipped with the code,
# so parsing stays fast and deterministic (no API or DB calls at top level).
from datetime import datetime
from airflow.sdk import DAG, get_parsing_context, task

TENANTS = {                       # in practice: yaml.safe_load(open(".../tenants.yaml"))
    "acme":   {"schedule": "0 2 * * *", "tables": ["orders", "invoices"]},
    "globex": {"schedule": "30 3 * * *", "tables": ["orders"]},
}


def build_dag(tenant: str, cfg: dict) -> DAG:
    # A plain builder function: arguments of an @dag-decorated function would
    # become DAG params instead of Python values.
    with DAG(dag_id=f"export_{tenant}", schedule=cfg["schedule"], start_date=datetime(2026, 1, 1),
             catchup=False, tags=["export", tenant]) as d:
        @task
        def export_table(table: str):
            print(f"exporting {tenant}.{table}")

        export_table.expand(table=cfg["tables"])
    return d


# A worker running one task sets the parsing context, so it can skip the other DAGs.
# The DAG processor leaves it unset and builds all of them.
current = get_parsing_context().dag_id
for tenant, cfg in TENANTS.items():
    if current is None or current == f"export_{tenant}":
        build_dag(tenant, cfg)       # a DAG created at module level is registered

# Runtime: Python 3.11, Apache Airflow 3.3 — check script outside the DAG folder
from airflow.dag_processing.dagbag import DagBag
bag = DagBag(dag_folder="dags")
for dag_id in sorted(bag.dag_ids):
    d = bag.get_dag(dag_id)
    print(dag_id, type(d.timetable).__name__, repr(d.timetable.expression), sorted(d.tags))
print("errors:", bag.import_errors)

export_acme CronTriggerTimetable '0 2 * * *' ['acme', 'export']
export_globex CronTriggerTimetable '30 3 * * *' ['export', 'globex']
errors: {}

Real-world use case

A B2B SaaS company ran one export DAG per customer, generated from a tenants.yaml file kept in the DAG repo. Each customer had its own schedule in its own time zone, and support staff could pause a single customer’s DAG. Inside each DAG, the tables were mapped, because the table list changed whenever a customer enabled a module.

Common mistakes and red flags

• Building the DAG list from a database query at parse time.
• Generating thousands of near-identical DAGs where one mapped DAG would do.
• Changing dag_ids on every deploy, for example by including a version or date in them.

Follow-up questions

• How do you test a factory? — Load the DagBag in CI, assert the expected dag_ids, and check that there are no import errors (see How do you write DAG integrity tests that catch broken DAGs before merge?).
• What happens when a tenant is removed from the YAML? — Its DAG stops being parsed and is marked stale. Its history stays in the DB until someone deletes it.

Detailed explanation

• Declare: Param(default, type=..., enum=..., minimum=..., maximum=..., description=...). Types follow JSON Schema: string, integer, number, boolean, array, object and null, plus formats such as date.
• Override: the trigger form, airflow dags trigger --conf '{...}', the REST API or TriggerDagRunOperator(conf=...). With [core] dag_run_conf_overrides_params = True (the default), conf values replace the defaults.
• Validate: values are checked against the schema when the run is created. A scheduled run uses the defaults, so every default must be valid.
• Read: params["region"] in TaskFlow, or {{ params.region }} in templates. Task-level params merge over DAG-level ones.

Params versus other inputs:

Good practice: give every param a safe default, so scheduled runs need no conf. Use enum for choices. Document params with description, which appears in the trigger form. Treat conf as untrusted user input: never paste it straight into SQL; bind it as a query parameter or check it against an allow-list.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import Param, dag, task


@dag(
    schedule=None,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    params={
        "region": Param("eu", type="string", enum=["eu", "us", "apac"]),
        "lookback_days": Param(1, type="integer", minimum=1, maximum=31),
        "dry_run": Param(False, type="boolean", description="Log the SQL only"),
        "tables": Param(["orders"], type="array", items={"type": "string"}),
    },
)
def reprocess_region():
    @task
    def plan(params=None):
        print(f"region={params['region']} lookback={params['lookback_days']} "
              f"dry_run={params['dry_run']} tables={params['tables']}")

    plan()


dag_obj = reprocess_region()

if __name__ == "__main__":
    print(dag_obj.test().state)                                    # defaults
    print(dag_obj.test(run_conf={"region": "us", "lookback_days": 7,
                                 "tables": ["orders", "refunds"]}).state)
    try:
        dag_obj.test(run_conf={"lookback_days": 90})
    except Exception as e:
        print(type(e).__name__, "->", str(e).splitlines()[0])

region=eu lookback=1 dry_run=False tables=['orders']
success
region=us lookback=7 dry_run=False tables=['orders', 'refunds']
success
ParamValidationError -> Invalid input for param lookback_days: 90 is greater than the maximum of 31

Real-world use case

A data team gave its support analysts a “reprocess region” DAG with no schedule. The analysts triggered it from the UI form and picked a region from a drop-down plus a lookback of up to 31 days. The schema blocked a 365-day request that would have overloaded the warehouse, and the engineers no longer had to run CLI commands for these requests.

Common mistakes and red flags

• Reading dag_run.conf["x"] directly with no default or validation.
• Formatting conf values straight into SQL strings.
• Using Variables as run inputs. A Variable is shared by every run.

Follow-up questions

• Can a scheduled run receive params? — Only the defaults. Params are meant for manual or API-triggered runs.
• How do you pass params to a downstream DAG? — TriggerDagRunOperator(conf={...}), which the target validates against its own params (see TriggerDagRunOperator, ExternalTaskSensor or assets: how do you choose for cross-DAG dependencies?).

Detailed explanation

The first run for "0 6 * * *" with start_date 2026-03-01 (output below):

Why this matters when you migrate: {{ ds }} and data_interval_* move between the two timetables. A task filtering on data_interval_start <= ts < data_interval_end would select nothing under a zero-length interval. Options:

• Set [scheduler] create_cron_data_intervals = True to keep the 2.x semantics for every cron-string DAG. Do it before upgrading.
• Pass an explicit CronDataIntervalTimetable(...) on the DAGs that depend on intervals.
• Or use CronTriggerTimetable(..., interval=timedelta(days=1)) and check your templates.

Built-in timetables to know: DeltaTriggerTimetable / DeltaDataIntervalTimetable, MultipleCronTriggerTimetable (several crons, one DAG), EventsTimetable (a fixed list of datetimes), AssetOrTimeSchedule, and the partition timetables from 3.2+ (see What is asset partitioning in Airflow 3.2+, and when would you use it?).

A custom timetable subclasses Timetable and implements next_dagrun_info and infer_manual_data_interval. It is registered through a plugin, because the scheduler must import it to deserialise DAGs. Typical cases are trading-day calendars and “business day N of the month”.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
# Compare how two timetables label and time the first run for "every day at 06:00".
from datetime import datetime, timedelta, timezone
from airflow.timetables.base import TimeRestriction
from airflow.timetables.interval import CronDataIntervalTimetable
from airflow.timetables.trigger import CronTriggerTimetable

start = datetime(2026, 3, 1, tzinfo=timezone.utc)
tables = {
    "CronTriggerTimetable (3.x default for cron strings)": CronTriggerTimetable("0 6 * * *", timezone="UTC"),
    "CronTriggerTimetable with interval=1 day": CronTriggerTimetable(
        "0 6 * * *", timezone="UTC", interval=timedelta(days=1)),
    "CronDataIntervalTimetable (2.x behaviour)": CronDataIntervalTimetable("0 6 * * *", timezone="UTC"),
}
for name, tt in tables.items():
    info = tt.next_dagrun_info(
        last_automated_data_interval=None,
        restriction=TimeRestriction(earliest=start, latest=None, catchup=True),
    )
    print(name)
    print("  logical_date:", info.logical_date)
    print("  interval:    ", info.data_interval.start, "->", info.data_interval.end)
    print("  runs at:     ", info.run_after)

CronTriggerTimetable (3.x default for cron strings)
  logical_date: 2026-03-01 06:00:00+00:00
  interval:     2026-03-01 06:00:00+00:00 -> 2026-03-01 06:00:00+00:00
  runs at:      2026-03-01 06:00:00+00:00
CronTriggerTimetable with interval=1 day
  logical_date: 2026-02-28 06:00:00+00:00
  interval:     2026-02-28 06:00:00+00:00 -> 2026-03-01 06:00:00+00:00
  runs at:      2026-03-01 06:00:00+00:00
CronDataIntervalTimetable (2.x behaviour)
  logical_date: 2026-03-01 06:00:00+00:00
  interval:     2026-03-01 06:00:00+00:00 -> 2026-03-02 06:00:00+00:00
  runs at:      2026-03-02 06:00:00+00:00

Real-world use case

After an upgrade to Airflow 3, a team’s hourly incremental loads started reporting zero rows. Their SQL filtered on data_interval_start/data_interval_end, and the new default timetable gave a zero-length interval. They switched those DAGs to an explicit CronDataIntervalTimetable and added a CI test that asserts the interval length of every interval-based DAG.

Common mistakes and red flags

• Saying “a daily DAG always runs for yesterday”. That depends on the timetable.
• Changing create_cron_data_intervals after Airflow 3 runs already exist. Logical dates shift, and runs can be duplicated or skipped.
• Writing a custom timetable without registering it in a plugin.

Follow-up questions

• Which one fits “run at 9:00 and use whatever is there”? — CronTriggerTimetable, the default.
• How does EventsTimetable help? — It schedules runs on an explicit list of dates, such as earnings dates, with no cron at all.

Detailed explanation

In 2.x, airflow dags backfill ran a loop in your terminal: closing the terminal killed it, and it ignored part of the scheduler’s limits. Airflow 3 moved backfills into the scheduler (AIP-78).

Backfill runs have run_type=backfill, appear in the grid, and can be paused or cancelled. A date that is currently running or queued is never duplicated. Backfills need a time-based schedule. For DAGs on partition timetables, the window is read as partition dates.

The DAG still has to be idempotent per interval (see How do you make Airflow tasks idempotent and safe to retry?, How do you design DAGs so any past interval can be reprocessed safely?), because a backfill simply replays history.

Example

# Runtime: Apache Airflow 3.3 CLI; DAG "daily_revenue" has schedule "0 3 * * *"
airflow backfill create --dag-id daily_revenue \
  --from-date 2026-03-01 --to-date 2026-03-05 \
  --reprocess-behavior failed --max-active-runs 2 --dry-run

Performing dry run of backfill.
Printing params:
    - dag_id = daily_revenue
    - from_date = 2026-03-01 00:00:00+00:00
    - to_date = 2026-03-05 00:00:00+00:00
    - max_active_runs = 2
    - reverse = False
    - dag_run_conf = None
    - reprocess_behavior = ReprocessBehavior.FAILED
    - run_on_latest_version = True
Runs to be attempted:
+---------------------------+-----------------+------------------+
| logical_date              | partition_key   | partition_date   |
+===========================+=================+==================+
| 2026-03-01 03:00:00+00:00 |                 |                  |
+---------------------------+-----------------+------------------+
| 2026-03-02 03:00:00+00:00 |                 |                  |
+---------------------------+-----------------+------------------+
| 2026-03-03 03:00:00+00:00 |                 |                  |
+---------------------------+-----------------+------------------+
| 2026-03-04 03:00:00+00:00 |                 |                  |
+---------------------------+-----------------+------------------+

The 5 March run is missing because the window ends at 00:00 that day. Use --to-date 2026-03-05T23:59 to include it.

Real-world use case

A currency-conversion bug inflated revenue for six weeks. After deploying the fix, the team ran a dry run, then --reprocess-behavior completed --max-active-runs 3 --run-backwards. The most recent weeks, which finance needed first, were corrected within hours, and the warehouse never ran more than three rebuilds at once.

Common mistakes and red flags

• Describing the 2.x airflow dags backfill CLI loop as current behaviour.
• Assuming the default (none) reprocesses dates that already ran. It only fills gaps.
• Backfilling a DAG whose tasks read “latest” data instead of their own interval.

Follow-up questions

• Catchup or backfill? — Catchup automatically fills missed intervals since start_date. A backfill is an explicit range that you start (see Airflow catchup, backfill, start_date and logical date).
• How do you protect production during a big backfill? — Use a low --max-active-runs, pools on the heavy tasks, and lower priority weights (see What are pools, and how do you use them to protect a shared system?, How do priority weights and weight_rule decide what runs first?).

In depth: What is the difference between catchup and backfill in Airflow 3?

Detailed explanation

What changed compared with 2.x:

• A logical date was always present and unique per DAG. With None, multiple runs no longer collide, which is why inference-style DAGs (AIP-83) became possible.
• Triggering with a future logical date is no longer allowed. Use logical_date=None instead.
• When the run has no logical date, context keys such as data_interval_start are missing from the context. The 3.0 release notes say reading them raises a KeyError.
• The conf context key and dag_run.external_trigger were removed. Use params and dag_run.run_type.

Defensive pattern: read dag_run.logical_date, fall back to dag_run.run_after or a param, and log which one you used. Or make the date a required Param on DAGs that are only triggered manually (see How do Params work, and how do you trigger a DAG with validated inputs?).

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime, timezone
from airflow.sdk import dag, task


@dag(schedule="0 4 * * *", start_date=datetime(2026, 1, 1), catchup=False)
def run_types_demo():
    @task
    def which_day(**context) -> str:
        dr = context["dag_run"]
        # logical_date is None for API/UI triggers without a date and for asset-triggered runs.
        # Fall back to run_after, which every run has, and say so in the log.
        anchor = dr.logical_date or dr.run_after
        print(f"run_type={dr.run_type} logical_date={dr.logical_date} "
              f"run_after={dr.run_after:%Y-%m-%d %H:%M} -> processing {anchor:%Y-%m-%d}")
        print("has data_interval_start:", "data_interval_start" in context)
        return anchor.strftime("%Y-%m-%d")

    which_day()


dag_obj = run_types_demo()

if __name__ == "__main__":
    dag_obj.test(logical_date=datetime(2026, 4, 2, 4, tzinfo=timezone.utc))   # like a scheduled/dated run
    dag_obj.test(logical_date=None, run_after=datetime(2026, 4, 9, 10, 30, tzinfo=timezone.utc))

run_type=DagRunType.MANUAL logical_date=2026-04-02 04:00:00+00:00 run_after=2026-10-17 13:45 -> processing 2026-04-02
has data_interval_start: True
run_type=DagRunType.MANUAL logical_date=None run_after=2026-04-09 10:30 -> processing 2026-04-09
has data_interval_start: False

Real-world use case

A fraud team’s scoring DAG was triggered through the REST API for every batch of flagged transactions, often dozens of times an hour. In 2.x each trigger needed a unique, made-up logical date. In 3.x they trigger without one and pass the batch id as a param. Each run reads its inputs from conf, and nothing in the DAG depends on dates.

Common mistakes and red flags

• Assuming every run has ds and a data interval.
• Inventing future logical dates to make runs unique.
• Using run_after as the data date for scheduled runs. It is when the run may start, not which data it covers.

Follow-up questions

• How do you trigger a dated rerun from the CLI? — airflow dags trigger <dag_id> --logical-date 2026-04-02T04:00:00+00:00.
• Which data interval does a manual run with a date get? — Whatever the timetable infers. Don’t assume it equals the logical date; check the context (see CronTriggerTimetable vs CronDataIntervalTimetable: what changed in Airflow 3, and when do you write a custom timetable?).

In depth: What are the logical date and data interval in Airflow, and what changed in Airflow 3?

Detailed explanation

• Identity. In Airflow 3, Asset(name=..., uri=...) can be identified by name, URI or both. The URI is normalised, so a trailing / is dropped. Pick one convention and use it everywhere.
• Producing. A task with outlets=[asset] emits an event when it succeeds. Use outlet_events[asset].extra = {...} to attach row counts, partitions or paths. Events can also come from the REST API, or from other DAGs through AssetAlias.
• Consuming. schedule=[a, b] means all of them. a & b is all, a | b is any, and the operators nest. Each consumer keeps a queue of the events received since its last run. Two producer events before one consumer run give one run that sees both events.
• Mixing with time. AssetOrTimeSchedule(timetable=..., assets=...) runs on a timer or on updates.
• What it doesn’t do. Airflow doesn’t check the data. An event means “the producer task succeeded”, not “the data is valid”. Put quality checks before the outlet task.

See TriggerDagRunOperator, ExternalTaskSensor or assets: how do you choose for cross-DAG dependencies? for the trade-offs and the site’s guide on cross-DAG dependencies.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import Asset, dag, task

orders = Asset(name="silver_orders", uri="s3://lake/silver/orders/")
customers = Asset(name="silver_customers", uri="s3://lake/silver/customers/")
fx_rates = Asset(name="fx_rates", uri="s3://lake/ref/fx_rates/")


@dag(schedule="@hourly", start_date=datetime(2026, 1, 1), catchup=False)
def load_orders():
    @task(outlets=[orders])          # success of this task emits an asset event
    def write_orders(*, outlet_events):
        outlet_events[orders].extra = {"rows": 1834, "partition": "2026-05-20T10"}

    write_orders()


# Runs when orders AND customers have both updated since the last run, OR fx_rates updated.
@dag(schedule=(orders & customers) | fx_rates, start_date=datetime(2026, 1, 1), catchup=False)
def build_revenue_mart():
    @task
    def build(*, triggering_asset_events):
        for asset, events in triggering_asset_events.items():
            print(asset.name, [e.extra for e in events])

    build()


producer = load_orders()
consumer = build_revenue_mart()

if __name__ == "__main__":
    print("consumer timetable:", type(consumer.timetable).__name__)
    def show(c):  # compact view of the condition tree
        return c.name if hasattr(c, "name") else f"{type(c).__name__}({', '.join(show(o) for o in c.objects)})"
    print("condition:", show(consumer.timetable.asset_condition))
    print("normalised uri:", orders.uri)
    print("producer run:", producer.test().state)
    from airflow.models.asset import AssetEvent
    from airflow.utils.session import create_session
    with create_session() as s:
        for ev in s.query(AssetEvent).order_by(AssetEvent.id.desc()).limit(1):
            print("event:", ev.asset.name, ev.extra, "from", ev.source_dag_id, ev.source_task_id)

consumer timetable: AssetTriggeredTimetable
condition: AssetAny(AssetAll(silver_orders, silver_customers), fx_rates)
normalised uri: s3://lake/silver/orders
producer run: success
event: silver_orders {'rows': 1834, 'partition': '2026-05-20T10'} from load_orders write_orders

The DB query at the end only inspects the result in this local test. Task code in Airflow 3 cannot query the metadata DB (see What are the Task SDK and the Task Execution API, and why do they matter?).

Real-world use case

A company replaced a 6 a.m. cron on its revenue mart, which ran whether or not the orders load had finished, with schedule=(orders & customers) | fx_rates. The mart now built minutes after both loads landed. The extra rebuild when FX rates were corrected happened automatically, and the empty-dashboard incidents on late-load days stopped.

Common mistakes and red flags

• Thinking an asset event proves the data is correct, or that Airflow watches the storage path. It only records that the producer task succeeded.
• Expecting one consumer run per producer event. Events are batched until the consumer runs.
• Using different URIs for the same data in different DAGs, such as with and without a trailing slash, or s3a:// versus s3://.

Follow-up questions

• Can an external system emit an event? — Yes, through the REST API’s asset-events endpoint, or event-driven with an AssetWatcher (see How do AssetWatcher and event-driven scheduling work?).
• What is AssetAlias? — A placeholder whose concrete asset is chosen at run time, for example when the output path depends on the data (see What do the @asset decorator, Metadata and AssetAlias add on top of plain asset outlets?).

Detailed explanation

When to use which:

• Use @asset when the team thinks in data products (“the daily_summary table”) and each product is one step. Use normal DAGs when a pipeline has many steps but publishes one result.
• Use AssetAlias when the set of outputs depends on the data. Defining a fixed asset for each possible output would be impractical.
• Asset events are lineage records too. The UI’s asset graph shows which DAG produced and consumed what, so use names people recognise.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from airflow.sdk import Asset, AssetAlias, Metadata, asset, dag, task


# @asset = one DAG + one task + one Asset, all named after the function.
@asset(uri="s3://lake/raw/orders", schedule="@daily")
def raw_orders(self):
    rows = 4210                                   # ... write the file ...
    yield Metadata(self, {"row_count": rows})     # becomes the asset event's extra


# Naming a parameter after another asset makes it an inlet; schedule on it to run after it.
@asset(uri="s3://lake/gold/daily_summary", schedule=raw_orders)
def daily_summary(self, context, raw_orders):
    last = context["inlet_events"][raw_orders][-1]   # most recent upstream event
    print("upstream rows:", last.extra["row_count"])


# AssetAlias: the concrete asset (here, which country partition) is only known at run time.
@dag(schedule=None)
def export_by_country():
    @task(outlets=[AssetAlias("country_exports")])
    def export(*, outlet_events):
        for country in ("de", "fr"):
            outlet_events[AssetAlias("country_exports")].add(
                Asset(name=f"export_{country}", uri=f"s3://exports/{country}/"), extra={"country": country})

    export()

export_by_country()


@dag(schedule=AssetAlias("country_exports"))     # runs when any asset resolved by the alias updates
def notify_partners():
    @task
    def notify(*, triggering_asset_events):
        print(sorted(a.name for a in triggering_asset_events))

    notify()

notify_partners()

# DagBag + dag.test() of raw_orders, daily_summary and export_by_country:
daily_summary      tasks=['daily_summary'] timetable=AssetTriggeredTimetable outlets=['daily_summary']
export_by_country  tasks=['export'] timetable=NullTimetable outlets=['country_exports']
notify_partners    tasks=['notify'] timetable=AssetTriggeredTimetable outlets=[]
raw_orders         tasks=['raw_orders'] timetable=CronTriggerTimetable outlets=['raw_orders']
raw_orders run: success
upstream rows: 4210
daily_summary run: success
export run: success
event: export_fr {'country': 'fr'}
event: export_de {'country': 'de'}

Real-world use case

A data-products team declared each gold table as an @asset. They yielded the row count and the max event time as Metadata. A freshness dashboard read these from the asset events API, so it could report “last updated, rows” for every product without scanning the warehouse.

Common mistakes and red flags

• Defining the same asset with slightly different names or URIs in two files. Use Asset.ref or a shared module.
• Putting large payloads in event extra. Like XCom, it belongs in the metadata DB only when small.
• Assuming @asset works for multi-step pipelines. It creates exactly one task; use @task with outlets in a normal DAG instead.

Follow-up questions

• How do classic operators attach metadata? — Through context["outlet_events"][asset].extra inside execute, or by subclassing the operator.
• Are asset events emitted on failure? — No. Only a successful task emits its outlets’ events.

Detailed explanation

This question is about configuring sensors. For poke versus reschedule versus deferrable, see Sensors vs deferrable operators and the site’s guide on how sensors should wait.

Things to know:

• timeout counts total time since the first check, and retries don’t reset it. In reschedule mode, execution_timeout doesn’t limit the overall wait either; timeout does.
• A sensor that returns a value saves a second lookup. Pass the discovered path forward with PokeReturnValue.
• Sensors fill pools like any other task. Hundreds of poke sensors can deadlock a pool.
• If the producer is inside Airflow, an asset is usually better than any sensor (see How does asset-aware scheduling work, and how do you combine asset conditions?).

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import PokeReturnValue, dag, task

ATTEMPTS = {"n": 0}            # stand-in for "is the partner manifest there yet?"


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def sensor_demo():
    @task.sensor(
        poke_interval=1,            # seconds between checks (use minutes in production)
        timeout=30,                 # total seconds before giving up
        mode="poke",                # "reschedule" frees the worker slot between pokes
        exponential_backoff=False,
        soft_fail=False,            # True = mark skipped instead of failed on timeout
    )
    def wait_for_manifest() -> PokeReturnValue:
        ATTEMPTS["n"] += 1
        found = ATTEMPTS["n"] >= 3
        print(f"poke {ATTEMPTS['n']}: found={found}")
        # is_done ends the wait; xcom_value is passed downstream like a return value
        return PokeReturnValue(is_done=found, xcom_value="s3://partner/2026-05-20/manifest.json" if found else None)

    @task
    def load(manifest: str):
        print("loading from", manifest)

    load(wait_for_manifest())


dag_obj = sensor_demo()

if __name__ == "__main__":
    print(dag_obj.test().state)

poke 1: found=False
poke 2: found=False
poke 3: found=True
loading from s3://partner/2026-05-20/manifest.json
success

The module-level counter only works because dag.test() runs every poke in one process. A real sensor must check outside state, because in reschedule mode each poke is a new process.

Real-world use case

A bank’s partner sent a manifest file some time between 01:00 and 05:00. The team used an S3 key sensor in deferrable mode with timeout=5*3600 and soft_fail=False. When the partner didn’t deliver, the sensor failed at 06:00 and paged the on-call engineer, instead of waiting quietly for the default seven days.

Common mistakes and red flags

• Leaving timeout at its default, so a missing file holds a slot or a deferral for a week.
• A poke_interval of a few seconds against a rate-limited API.
• Keeping state in Python globals inside a reschedule-mode sensor.

Follow-up questions

• What does up_for_reschedule mean? — A reschedule-mode sensor between checks. It holds no slot.
• How do you make sensors deferrable by default? — Set [operators] default_deferrable = True for operators and sensors that support it.

Detailed explanation

For a three-task chain extract >> clean >> load where load has priority_weight=10 and the others 1 (output below):

Points to make:

• Weights only matter when something is full: a pool, parallelism or executor slots. With spare capacity, everything ready runs.
• Weights are compared across DAGs. Use weights to put the CFO dashboard’s DAG ahead of an exploratory backfill.
• upstream is good for long DAGs, so half-finished runs don’t wait behind brand-new ones. That helps especially during catchup or backfills.
• absolute is cheap and predictable on very large DAGs.
• A custom PriorityWeightStrategy, registered through a plugin, can compute weights from task attributes.
• Weight doesn’t pre-empt anything. A running low-priority task keeps its slot.

Example

# Runtime: Python 3.11, Apache Airflow 3.3
from datetime import datetime
from airflow.sdk import DAG, WeightRule
from airflow.providers.standard.operators.empty import EmptyOperator

dags = {}
for rule in (WeightRule.DOWNSTREAM, WeightRule.UPSTREAM, WeightRule.ABSOLUTE):
    with DAG(f"weights_{rule.value}", schedule=None, start_date=datetime(2026, 1, 1),
             default_args={"weight_rule": rule}) as d:
        extract = EmptyOperator(task_id="extract")
        clean = EmptyOperator(task_id="clean")
        load = EmptyOperator(task_id="load", priority_weight=10)   # own weight 10, others 1
        extract >> clean >> load
    dags[rule.value] = d

if __name__ == "__main__":
    for name, d in dags.items():
        run = d.test()
        # the effective weight the scheduler stored on each task instance
        print(f"{name:10}", {ti.task_id: ti.priority_weight for ti in
                             sorted(run.get_task_instances(), key=lambda t: ["extract", "clean", "load"].index(t.task_id))})

downstream {'extract': 12, 'clean': 11, 'load': 10}
upstream   {'extract': 1, 'clean': 2, 'load': 12}
absolute   {'extract': 1, 'clean': 1, 'load': 10}

Real-world use case

During a 90-day backfill, the backfill’s runs crowded out the daily production loads in a shared pool. The team gave production DAGs priority_weight=100 with weight_rule="absolute" and left the backfill at 1. Production tasks got the next free slot every time, and the backfill used whatever capacity was left.

Common mistakes and red flags

• Expecting priority to speed things up when there is spare capacity.
• Raising weights everywhere (“everything is priority 100”), which changes nothing.
• Not knowing the default rule adds up downstream weights, so long DAGs win over short ones.

Follow-up questions

• Where is weight applied — the scheduler or the executor? — The scheduler orders what it queues. Celery queues and Kubernetes scheduling add their own ordering after that.
• How do you give a whole DAG priority? — default_args={"priority_weight": ..., "weight_rule": "absolute"}.

Detailed explanation

How retry policies work: rules are checked in order against the exception, using isinstance by default, and the first match wins. An action of RETRY can set its own delay but never goes beyond retries. FAIL stops now. DEFAULT falls back to the normal settings, and so does an exception no rule matches. The policy runs in the worker, and its decision is written to the task log. A subclass can inspect HTTP status codes, Retry-After headers or dag_run.run_type.

Rules of thumb:

• Retry transient errors such as timeouts, 429s, 5xx responses and lost connections. Fail fast on bad credentials, 4xx responses, schema errors and bad input.
• Retries are only safe if the task is idempotent (see How do you make Airflow tasks idempotent and safe to retry? and the site’s guide on retries and idempotency).
• Always set execution_timeout on tasks that call networks. A hung socket otherwise holds a slot until the heartbeat timeout fires (see What are zombie tasks (heartbeat timeouts), and how do you debug them?).

Example

# Runtime: Python 3.11, Apache Airflow 3.3 (retry policies are new in 3.3)
from datetime import datetime, timedelta
from airflow.sdk import ExceptionRetryPolicy, RetryAction, RetryRule, dag, task

CALLS = {"flaky": 0}

# First matching rule wins; anything unmatched falls back to retries/retry_delay.
API_POLICY = ExceptionRetryPolicy(rules=[
    RetryRule(exception=PermissionError, action=RetryAction.FAIL, reason="bad credentials, retrying won't help"),
    RetryRule(exception=ConnectionError, action=RetryAction.RETRY, retry_delay=timedelta(seconds=1)),
])


@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False,
     default_args={
         "retries": 3,
         "retry_delay": timedelta(seconds=1),
         "retry_exponential_backoff": 2.0,          # numeric multiplier since 3.2 (True = 2)
         "max_retry_delay": timedelta(minutes=30),
         "execution_timeout": timedelta(minutes=10),  # per try; kills a hung call
         "retry_policy": API_POLICY,
     })
def retry_demo():
    @task
    def flaky_api():
        CALLS["flaky"] += 1
        if CALLS["flaky"] < 3:
            raise ConnectionError(f"timeout on attempt {CALLS['flaky']}")
        print("succeeded on attempt", CALLS["flaky"])

    @task
    def bad_credentials():
        raise PermissionError("token rejected")

    flaky_api()
    bad_credentials()


dag_obj = retry_demo()

if __name__ == "__main__":
    run = dag_obj.test()
    for ti in sorted(run.get_task_instances(), key=lambda t: t.task_id):
        print(f"{ti.task_id:16} state={ti.state} try_number={ti.try_number} max_tries={ti.max_tries}")