Apache Airflow Reference

Workflows as code: architecture, DAGs with the TaskFlow API, scheduling and data intervals, operators and sensors, XCom, branching, dynamic tasks, assets, Spark/Databricks/Glue orchestration, testing and production.

1. What Apache Airflow Is

Apache Airflow is an open-source platform to author, schedule and monitor workflows as code. A workflow is a DAG (directed acyclic graph) of tasks written in Python: Airflow decides when each run should happen, runs tasks in dependency order on workers, retries failures, and shows everything in a web UI. Airflow orchestrates: the heavy processing usually happens elsewhere (Spark, Databricks, Glue, a warehouse, dbt).

This guide uses Airflow 3 syntax (released 2025). Where Airflow 2.x differs, it is called out. MWAA, Cloud Composer and many companies still run 2.x, so you will meet both.

2. Architecture

            ┌────────────── metadata database (Postgres / MySQL) ──────────────┐
            │  DAG runs, task states, connections, variables, XComs, history    │
            └───────▲───────────────▲───────────────▲───────────────▲───────────┘
                    │               │               │               │
  dags/ folder ─► DAG processor   Scheduler      API server       Triggerer
  (Python files)  (parses DAG     (decides what  (web UI + REST   (async waits for
                   files)          runs, queues   API; workers    deferrable tasks)
                                   tasks)         report through it)
                                      │
                                      ▼  executor
                               Workers (Celery / Kubernetes pods / local processes)

3. Your First DAG (TaskFlow API)

The TaskFlow API turns Python functions into tasks with @task. Return values are passed between tasks automatically (through XCom), and the dependencies follow from how the functions call each other.

from datetime import datetime, timedelta
from airflow.sdk import dag, task  # Airflow 2.x: from airflow.decorators import dag, task

@dag(
    schedule="0 2 * * *",  # 02:00 every day (Airflow 2.x also accepted schedule_interval)
    start_date=datetime(2026, 10, 1),
    catchup=False,  # don't create runs for every missed interval
    default_args={"owner": "data-platform", "retries": 2, "retry_delay": timedelta(minutes=5)},
    tags=["fraud", "daily"],
)
def daily_transactions():

    @task
    def extract(ds=None) -> str:  # ds = the run's date (YYYY-MM-DD), injected from the context
        return f"s3://acme-raw/transactions/{ds}/"

    @task
    def validate(path: str) -> int:
        rows = count_rows(path)              # your helper
        if rows == 0:
            raise ValueError(f"no data in {path}")  # fails the task → retries → alert
        return rows

    @task
    def load(path: str, rows: int) -> None:
        print(f"loading {rows} rows from {path} into silver.transactions")

    path = extract()
    load(path, validate(path))  # dependencies: extract → validate → load

daily_transactions()

Put the file in the dags/ folder; the DAG processor picks it up, and it appears in the UI as daily_transactions.

The same DAG with classic operators

from datetime import datetime
from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator  # Airflow 2.x: airflow.operators.bash
from airflow.providers.standard.operators.empty import EmptyOperator

with DAG(dag_id="spark_daily", schedule="@daily",
         start_date=datetime(2026, 10, 1), catchup=False) as dag:

    start = EmptyOperator(task_id="start")
    enrich = BashOperator(
        task_id="spark_enrich",
        bash_command="spark-submit jobs/enrich.py --run-date {{ ds }}",  # Jinja template
    )
    publish = BashOperator(task_id="publish", bash_command="python jobs/publish.py {{ ds }}")
    done = EmptyOperator(task_id="done")

    start >> enrich >> publish >> done  # bit-shift operators set dependencies
    # fan-out / fan-in:  start >> [a, b, c] >> done

4. Scheduling, Data Intervals & Catchup

Each DAG run covers a data interval. A run is created when its interval ends, because only then is the data for that interval complete. This is the most common source of confusion in Airflow interviews.

schedule = "0 2 * * *"   (daily at 02:00)

data interval:   2026-10-25 02:00  ──────────────►  2026-10-26 02:00
logical date:    2026-10-25 02:00  (= start of the interval; {{ ds }} = "2026-10-25")
run starts:                                          2026-10-26 02:00  (after the interval closes)

Templates and run-specific values

Use the run's logical date, never "today", so reruns and backfills process the right data:

{{ ds }}                      → 2026-10-25                  (logical date)
{{ ds_nodash }}               → 20261025
{{ data_interval_start }}     → 2026-10-25T02:00:00+00:00
{{ data_interval_end }}       → 2026-10-26T02:00:00+00:00
{{ macros.ds_add(ds, -1) }}   → 2026-10-24
{{ params.env }}              → value from the DAG's params
{{ var.value.raw_bucket }}    → an Airflow Variable
{{ conn.my_postgres.host }}   → a field of a connection

# in TaskFlow code, read the context directly:
from airflow.sdk import get_current_context
ctx = get_current_context()
start = ctx["data_interval_start"]

Airflow 3 removed execution_date: use logical_date (or data_interval_start). Templates are only rendered in fields an operator lists as templated (such as bash_command, sql, and many *_args parameters).

5. Operators, Sensors & Hooks

Hooks are the reusable clients behind operators (e.g. S3Hook, PostgresHook). Use a hook inside your own task when no operator fits; it reads credentials from an Airflow connection.

Sensors: wait for data, efficiently

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

wait_for_file = S3KeySensor(
    task_id="wait_for_success_marker",
    bucket_key="s3://acme-raw/landing/transactions/{{ ds }}/_SUCCESS",
    aws_conn_id="aws_default",
    deferrable=True,  # hand the wait to the triggerer: no worker slot used
    poke_interval=300,            # check every 5 minutes
    timeout=6 * 60 * 60,  # give up (fail) after 6 hours → alert
)
wait_for_file >> enrich

6. Passing Data Between Tasks (XCom)

XCom ("cross-communication") stores small values in the metadata DB so one task can pass them to another. TaskFlow return values are XComs automatically.

@task
def find_partitions(ds=None) -> list[str]:
    return [f"event_date={ds}/channel={c}" for c in ("UPI", "CARD", "ATM")]  # small: fine

@task
def report(partitions: list[str]) -> None:
    print(len(partitions), "partitions processed")

report(find_partitions())

# classic operators: read another task's XCom in a template
#   "{{ ti.xcom_pull(task_ids='find_partitions') }}"

Pass references, not data: an S3 path, a table name, a row count. Never pass DataFrames or large files through XCom; it lives in the metadata database. Large values need a custom object-storage XCom backend, and even then should be avoided.

7. Branching, Trigger Rules & Dynamic Tasks

Branching

@task.branch
def choose_load_type(ds=None) -> str:
    from datetime import date
    day = date.fromisoformat(ds)
    return "full_reload" if day.day == 1 else "incremental_load"  # task_id to follow

@task
def full_reload(): ...

@task
def incremental_load(): ...

@task(trigger_rule="none_failed_min_one_success")  # run after whichever branch ran
def publish(): ...

choose_load_type() >> [full_reload(), incremental_load()] >> publish()

Dynamic task mapping

Create one task instance per item at runtime, without knowing the count in advance:

@task
def list_new_files(ds=None) -> list[str]:
    return s3_list(f"s3://acme-raw/landing/{ds}/")  # e.g. 37 files today, 52 tomorrow

@task(max_active_tis_per_dag=8)  # at most 8 in parallel
def load_file(path: str) -> int:
    return copy_into_bronze(path)

@task
def total(counts: list[int]) -> None:
    print("rows loaded:", sum(counts))

total(load_file.expand(path=list_new_files()))  # one load_file per file

.partial(conn_id="x").expand(path=…) fixes some arguments and maps others; expand_kwargs maps over a list of argument dicts. Each mapped instance retries independently.

Task groups

from airflow.sdk import task_group

@task_group
def bronze_to_silver(table: str):
    validate_bronze(table) >> merge_into_silver(table)  # tasks defined elsewhere with @task

for t in ["transactions", "customers", "accounts"]:
    bronze_to_silver.override(group_id=f"silver_{t}")(t)

Task groups organise the UI. SubDAGs were removed in Airflow 3; use task groups, or separate DAGs connected with assets.

8. Data-Aware Scheduling with Assets

Instead of guessing times ("silver usually finishes by 03:00, so start gold at 03:30"), let a DAG run when the data it needs has been updated. A task declares the asset it produces; consumer DAGs schedule on it.

from airflow.sdk import Asset, dag, task  # Airflow 2.x: from airflow.datasets import Dataset

silver_orders = Asset("s3://acme-lake/silver/orders")

@dag(schedule="@hourly", start_date=datetime(2026, 10, 1), catchup=False)
def build_silver_orders():
    @task(outlets=[silver_orders])  # marks the asset updated when this task succeeds
    def merge_orders():
        ...
    merge_orders()

@dag(schedule=[silver_orders], start_date=datetime(2026, 10, 1), catchup=False)
def build_gold_sales():  # runs after each successful silver update
    @task
    def aggregate():
        ...
    aggregate()

build_silver_orders()
build_gold_sales()

Schedule on several assets (schedule=[a, b] waits for both) or combine with time using asset-or-time schedules. Assets also give cross-DAG lineage in the UI. In Airflow 2.4–2.10 the same feature is called Datasets.

9. Orchestrating Spark, Databricks, Glue & dbt

from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator
from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
from airflow.providers.amazon.aws.operators.emr import EmrServerlessStartJobOperator

run_databricks = DatabricksRunNowOperator(
    task_id="silver_merge",
    databricks_conn_id="databricks_default",
    job_id=123456,  # an existing Databricks job (Asset Bundle)
    job_parameters={"run_date": "{{ ds }}"},
    deferrable=True,
)

run_glue = GlueJobOperator(
    task_id="raw_to_curated",
    job_name="raw_to_curated",
    script_args={"--run_date": "{{ ds }}"},
    aws_conn_id="aws_default",
    wait_for_completion=True,
)

run_emr = EmrServerlessStartJobOperator(
    task_id="enrich_on_emr",
    application_id="00fabcdexample",
    execution_role_arn="arn:aws:iam::123456789012:role/emr-serverless-job",
    job_driver={"sparkSubmit": {
        "entryPoint": "s3://acme-code/jobs/enrich.py",
        "entryPointArguments": ["--run-date", "{{ ds }}"],
    }},
)

# dbt: run models for the day, then test them
dbt_run = BashOperator(task_id="dbt_run",
                       bash_command="cd /opt/dbt && dbt build --select tag:daily --vars '{run_date: {{ ds }}}'")

• Keep transformation logic in the engine (Spark job, Databricks job, dbt project) and let Airflow pass parameters, order the steps, retry and alert. This keeps DAGs thin and testable.
• Prefer deferrable operators for long-running external jobs: the worker slot is freed while the job runs.
• The Cosmos package (by Astronomer) turns each dbt model into its own Airflow task, for per-model retries and visibility.

10. Connections, Variables & Secrets

Connections hold credentials and endpoints (host, login, password, extras) referenced by ID such as aws_default. Variables hold small configuration values. Both can live in the metadata DB, in environment variables, or (best) in a secrets backend.

# environment variables (handy in containers and CI)
export AIRFLOW_CONN_WAREHOUSE='postgresql://etl_user:s3cret@warehouse.internal:5432/analytics'
export AIRFLOW_VAR_RAW_BUCKET='acme-raw'

# CLI
airflow connections add aws_default --conn-type aws --conn-extra '{"region_name": "ap-south-1"}'
airflow variables set env prod

from airflow.sdk import Variable  # Airflow 2.x: from airflow.models import Variable
from airflow.providers.postgres.hooks.postgres import PostgresHook

@task
def row_count(table: str) -> int:
    env = Variable.get("env", default="dev")  # read inside tasks, not at the top of the file
    hook = PostgresHook(postgres_conn_id="warehouse")
    return hook.get_first(f"SELECT COUNT(*) FROM {env}_{table}")[0]

• Configure a secrets backend (AWS Secrets Manager / SSM, Azure Key Vault, GCP Secret Manager, HashiCorp Vault) so credentials never sit in the DB or code.
• On AWS, prefer IAM roles for workers (no keys in the connection at all).
• Avoid Variable.get at the top level of a DAG file: it runs on every parse and hammers the database. Use it inside tasks or through templates.

11. Retries, Alerts & Resource Control

from datetime import timedelta

def notify_failure(context):
    ti = context["ti"]
    send_slack(f":red_circle: {ti.dag_id}.{ti.task_id} failed for {context['ds']} — {ti.log_url}")

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(hours=2),  # kill tasks that hang
    "on_failure_callback": notify_failure,
}

12. Testing & Debugging DAGs

# tests/test_dags.py: run in CI on every pull request
from airflow.models import DagBag

def test_dags_import_cleanly():
    bag = DagBag(dag_folder="dags/", include_examples=False)
    assert bag.import_errors == {}, bag.import_errors

def test_daily_transactions_structure():
    dag = DagBag(dag_folder="dags/", include_examples=False).get_dag("daily_transactions")
    assert dag is not None
    assert set(dag.task_ids) == {"extract", "validate", "load"}
    assert dag.default_args["retries"] == 2

airflow dags list
airflow dags list-import-errors  # syntax / import problems in DAG files
airflow dags test daily_transactions 2026-10-25  # run a whole DAG locally, no scheduler
airflow tasks test daily_transactions validate 2026-10-25  # run one task, no state saved
airflow dags trigger daily_transactions -c '{"full_reload": true}'
airflow tasks clear daily_transactions -t validate -s 2026-10-20 -e 2026-10-25  # rerun a range
airflow dags pause daily_transactions / airflow dags unpause daily_transactions

• Keep business logic in plain Python modules with ordinary unit tests; DAG files should mostly wire tasks together.
• Calling dag.test() at the bottom of a DAG file lets you run and debug it in your IDE.
• For local development use the official Docker Compose setup or the Astro CLI.

13. Running Airflow in Production

# MWAA deployment: everything is files in S3
aws s3 sync dags/ s3://acme-mwaa-prod/dags/ --delete
aws s3 cp requirements.txt s3://acme-mwaa-prod/requirements.txt  # pin provider versions
# then update the environment if requirements or plugins changed
aws mwaa update-environment --name acme-prod \
    --requirements-s3-object-version <version-id>

• Deploy DAGs through CI/CD (tests → sync to the DAGs bucket or image), never by editing files on servers.
• Pin Airflow and provider versions with the official constraints file; upgrade in a test environment first.
• Ship task logs to remote storage (S3, GCS, Azure Blob) and monitor scheduler health, queue length and task durations.

14. Best Practices

• Idempotent tasks: running a task twice for the same interval gives the same result (overwrite the partition or MERGE on keys, don't blindly append). Retries and backfills depend on this.
• Use the logical date ({{ ds }} / data_interval_start) for what to process, never datetime.now().
• Keep the top level of DAG files light: no database calls, API calls or heavy imports outside tasks. Files are parsed every few seconds or minutes.
• Atomic, focused tasks: one logical step per task, so retries redo only what failed.
• Small XComs: pass paths and IDs, not data.
• Static start_date and deliberate catchup, and a sensible max_active_runs.
• Orchestrate, don't compute: push heavy work to Spark, Databricks, the warehouse or Kubernetes pods.
• Alert with context: DAG, task, date and log link in every failure message, with an owner for every DAG (tags, owner field).

15. Airflow vs Other Orchestrators

16. Common Interview Questions