Airflow DAG Authoring cheat sheet

A DAG is a Python file that instantiates airflow.DAG (or uses @dag) with a unique dag_id; Airflow parses the file to discover DAGs and tasks.

Core DAG arguments: dag_id (unique string), schedule (cron, timedelta, timetable, or None), start_date, catchup, default_args, tags, max_active_runs, max_active_tasks.

start_date must be a timezone-aware datetime (pendulum); naive datetimes raise errors or are coerced to the scheduler timezone.

Tasks belong to a DAG via dag=dag on the operator, or via with DAG(...) as dag: context manager; task_id must be unique within a DAG.

default_args is a dict merged into every operator's kwargs; explicit operator kwargs override default_args, and default_args does not apply to DAG-level settings except those passed through.

DAG files live in the dags_folder; the scheduler parses them on a cadence (min_file_process_interval) and serializes DAG structure to the metadata DB.

DAG-level params (params={...}) supply default values accessible via context['params'] and are overridable at trigger time with conf.

doc_md, description, and tags improve UI discoverability; owner and retries are conventionally set in default_args.

A DAG with no tasks or a task_id collision fails to import; import errors appear in the UI and in scheduler logs.

Operators are templates for a single task; instantiating one creates a task node. Base classes: BaseOperator, and the taskflow @task decorator which wraps a Python function.

Common providers: BashOperator (bash_command), PythonOperator (python_callable), SQLExecuteQueryOperator, and EmptyOperator for flow control.

Operator arguments flow into the templated fields declared in template_fields; only those fields render Jinja at runtime.

Sensor operators (e.g., FileSensor, ExternalTaskSensor) wait for a condition; they support poke vs reschedule mode and timeout.

Taskflow (@task) infers dependencies from function arguments and returns; mixing taskflow with classic operators is supported via XComArg.

The operator's execute(context) method does the work; pre_execute/post_execute wrap it, and on_kill handles termination.

do_x3d (deprecated) is superseded by partial() and expand(); use functools.partial or operator.partial to freeze arguments.

Trigger rules (all_success, all_failed, one_success, none_failed, etc.) are set per task via trigger_rule.

Resources (pool, pool_slots, queue) constrain where and how many task instances run at once.

Bitshift operators: task_a >> task_b (a before b), task_a << task_b (b before a), and list forms [a, b] >> c fan-in, a >> [b, c] fan-out.

set_upstream/set_downstream are the method equivalents of << and >>; set_dependency is the older alias.

cross_downstream([a, b], [c, d]) creates all-to-all edges; chain(a, b, c) and chain(*groups) linearize sequences.

TaskGroup groups tasks in the UI and namespaces task_ids as group_id.task_id; edges can cross group boundaries.

trigger_rule decides when a task runs: all_success (default), all_failed, all_done, one_success, one_failed, none_failed, none_failed_min_one_success, none_skipped, always.

A task that is skipped propagates skip downstream under all_success; use none_failed_min_one_success for join nodes that tolerate skips.

Branching: BranchPythonOperator returns the task_id(s) to follow; all non-selected branches are skipped, and the join task needs an appropriate trigger_rule (e.g., none_failed_min_one_success).

ShortCircuitOperator skips all downstream tasks when its callable returns False; it does not raise.

Dynamic mapping: task.expand(arg=[...]) creates one mapped instance per element; task.partial freezes non-mapped args; mapped tasks can be followed by a reducing task.

The scheduler creates a DAG run for each interval defined by schedule and start_date; with catchup=True it backfills every interval since start_date.

A DAG run's logical date (formerly execution_date) is the start of the interval it covers, not the time it actually ran; use data_interval_start and data_interval_end for the covered window.

For a cron schedule, data_interval_start = the cron tick, data_interval_end = the next tick; a task runs after the interval closes (unless using a custom timetable).

schedule accepts cron strings, timedelta, @daily/@hourly presets, timetables, or None (manual trigger only); schedule_interval is the legacy name.

catchup=False makes the scheduler skip missed intervals and only run the most recent; max_active_runs caps concurrent runs per DAG.

Timezones: use pendulum.timezone('UTC') or zoneinfo; DAGs run in the scheduler/airflow timezone (default UTC), and cron is evaluated in that timezone.

depends_on_past=True blocks a task instance until the same task succeeded in the previous run; wait_for_downstream extends this to downstream tasks.

end_date stops scheduling new runs; a DAG with an end_date in the past will not create new runs.

The scheduler uses a heartbeat and parses files at min_file_process_interval; the DAG's schedule is stored in serialized form, so changing it requires a parse cycle.

XComs store small values in the metadata DB keyed by (dag_id, task_id, run_id, key); xcom_push(key, value) and xcom_pull(task_ids=..., key=...) are the APIs.

A task's return value is automatically pushed under key 'return'; taskflow functions receive pulled XComs as arguments when dependencies exist.

XCom values must be JSON-serializable and small; large payloads should go to object storage and pass a URI.

Templating: fields listed in template_fields are rendered with Jinja2 using context variables (ds, ds_nodash, ts, ts_nodash, data_interval_start, data_interval_end, params, var, conn, ti, task, dag).

{{ ds }} is the logical date in YYYY-MM-DD; {{ ds_nodash }} is YYYYMMDD; {{ ts }} is ISO-8601; {{ prev_ds }} / {{ next_ds }} refer to adjacent intervals.

{{ var.value.my_var }} reads an Airflow Variable; {{ conn.my_conn.host }} reads a Connection field; {{ params.x }} reads DAG/task params.

Custom templating: add template_fields = ('my_arg',) to a custom operator; use render_template_as_native_obj=True on the DAG to keep rendered types (dict/list) native.

XComArg (task.output) is the programmatic handle for passing output between tasks and into expand(); it does not require a manual xcom_pull.

XComs are pulled at runtime, not parse time; a downstream task only sees upstream XComs after the upstream task succeeds.

Connections store host, schema, login, password, port, and extra; conn_id is the lookup key and is referenced by hooks and operators.

Hooks (BaseHook subclasses) read a Connection and expose a client (e.g., PostgresHook, S3Hook); operators typically accept a conn_id and build a hook internally.

get_connection(conn_id) resolves from the metadata DB, environment variable AIRFLOW_CONN_<CONN_ID>, or secrets backend, in that precedence order.

Variables are key-value pairs stored in the metadata DB, read via Variable.get('key') or the templated {{ var.value.key }}; JSON values can be parsed with deserialize_json=True.

Environment overrides: AIRFLOW_VAR_ sets a Variable; AIRFLOW_CONN_<CONN_ID> (URI form) sets a Connection.

Secrets backends (e.g., AWS Secrets Manager, Vault, GCP Secret Manager) are configured via secrets_backend in airflow.cfg and are queried after env vars.

Avoid calling Variable.get() at DAG parse time; it hits the DB on every parse and slows scheduling. Use templated var.value or pass via params.

Connections and Variables are encrypted at rest when a Fernet key is set; without it, secrets are stored in plaintext.

conn_id naming conventions and pools (pool, pool_slots) belong to the same operational configuration layer; pools limit concurrency per resource.

Dynamic task mapping: task.expand(x=[1,2,3]) creates one mapped task instance per element at runtime; the number of instances is determined when the upstream produces the list.

task.partial(fixed_arg=...) freezes arguments so only the expanded argument varies; combine partial and expand on the same task.

The mapped argument can be an XComArg (upstream task output), a list, or a dict; expand_kwargs maps over dicts to produce keyword arguments.

A mapped task's output is a list of results; a downstream reduce task receives it as a list (or use a normal task as the reducer).

max_mapped_tasks (DAG-level) caps the number of mapped instances; exceeding it fails the task.

Dynamic DAG generation: loop over a config list to create multiple DAGs in one file, or use a factory function returning a DAG; keep dag_id unique.

Avoid heavy work (API calls, DB queries) at module level in DAG files; it runs on every parse. Cache results in a Variable or a separate process.

TaskGroups can be mapped (expand a group) to create parallel subgraphs; the group_id must be unique per expansion.

Mapped tasks inherit the operator's template_fields and trigger_rule; each instance gets its own task_id suffix based on the map index.

retries (int) and retry_delay (timedelta) control automatic retries; retry_exponential_backoff=True increases delay exponentially, capped by max_retry_delay.

on_failure_callback, on_success_callback, on_retry_callback, and on_skipped_callback receive the context dict; callbacks run in the scheduler/worker, so keep them lightweight.

The SLA is set via sla=timedelta(...) on a task; if the task exceeds it, an sla_miss_callback fires (DAG-level) and an event is logged; SLAs do not stop the task.

Airflow 2.6+ introduced the Deadline Alerts / SLA replacement (deadline=...), which can trigger a callback and optionally mark the task failed.

exit codes: a Python exception marks the task failed; BashOperator fails on a non-zero exit code; on_kill handles SIGTERM during task termination.

AirflowException raised in a callable fails the task; AirflowSkipException marks it skipped; AirflowFailException fails immediately without retries.

emulate_default_retries and the default trigger_rule interact: a failed task with retries left stays in up_for_retry, not failed.

Email alerts: email_on_failure, email_on_retry, and email=... on the operator, or a default_args entry; requires an SMTP connection and email backend configured.

Task timeout: execution_timeout kills a running task after the limit; dagrun_timeout limits the whole DAG run.

airflow dags test <dag_id> <logical_date> runs a DAG locally without the scheduler; airflow tasks test <dag_id> <task_id> <logical_date> runs a single task.

The DagBag class loads DAG files for validation: DagBag(dag_folder=..., include_examples=False) then assert dag_id in bag.dags.

Unit tests: call the callable directly, or use task.render_template / ti.xcom_push with a mock context; pytest with airflow.utils.db.clear_db for isolation.

Common failure causes: import errors (syntax, missing provider), task_id collisions, naive start_date, and circular dependencies.

Debug UI: the Graph view shows task states; the Grid view shows runs; the task instance log is in the Logs tab and on the worker.

Performance: keep DAG files small, avoid top-level DB/API calls, reduce parse frequency via min_file_process_interval, and use pools to limit load.

Airflow 2.x uses the scheduler's serialized DAG; the parsing process runs separately, so heavy imports slow the scheduler, not the workers.

Use airflow tasks states-for-dag-run and airflow dags list-runs to inspect state from the CLI; airflow dags report summarizes DAG health.

Testing connections: airflow connections test <conn_id> verifies a connection without running a DAG.

One DAG per logical workflow; keep DAGs idempotent and deterministic so reruns produce the same result.

Use default_args for retries, owner, and email; override only where needed; keep DAG-level defaults in a shared module.

Prefer taskflow (@task) for Python logic and classic operators for integrations; avoid putting business logic in the DAG file's top level.

Set catchup=False for most new DAGs unless backfill is required; use max_active_runs to bound concurrency.

Name tasks clearly with verb-noun task_ids; group related tasks in TaskGroups; add tags and doc_md for discoverability.

Avoid XCom for large data; pass URIs or use a shared storage layer; keep variables and connections out of parse-time code.

Use pools to limit concurrency against external systems; set execution_timeout to prevent hung tasks.

Version-control DAGs, review changes, and run CI checks (DagBag import, unit tests) before deploying; keep provider versions pinned.

Delete or archive stale DAGs; unused DAGs still consume scheduler parse time and clutter the UI.

A DAG is a Python file that instantiates airflow.DAG (or uses @dag) with a unique dag_id; Airflow parses the file to discover DAGs and tasks.

Core DAG arguments: dag_id (unique string), schedule (cron, timedelta, timetable, or None), start_date, catchup, default_args, tags, max_active_runs, max_active_tasks.

start_date must be a timezone-aware datetime (pendulum); naive datetimes raise errors or are coerced to the scheduler timezone.

Tasks belong to a DAG via dag=dag on the operator, or via with DAG(...) as dag: context manager; task_id must be unique within a DAG.

default_args is a dict merged into every operator's kwargs; explicit operator kwargs override default_args, and default_args does not apply to DAG-level settings except those passed through.

DAG files live in the dags_folder; the scheduler parses them on a cadence (min_file_process_interval) and serializes DAG structure to the metadata DB.

DAG-level params (params={...}) supply default values accessible via context['params'] and are overridable at trigger time with conf.

doc_md, description, and tags improve UI discoverability; owner and retries are conventionally set in default_args.

A DAG with no tasks or a task_id collision fails to import; import errors appear in the UI and in scheduler logs.

Operators are templates for a single task; instantiating one creates a task node. Base classes: BaseOperator, and the taskflow @task decorator which wraps a Python function.

Common providers: BashOperator (bash_command), PythonOperator (python_callable), SQLExecuteQueryOperator, and EmptyOperator for flow control.

Operator arguments flow into the templated fields declared in template_fields; only those fields render Jinja at runtime.

Sensor operators (e.g., FileSensor, ExternalTaskSensor) wait for a condition; they support poke vs reschedule mode and timeout.

Taskflow (@task) infers dependencies from function arguments and returns; mixing taskflow with classic operators is supported via XComArg.

The operator's execute(context) method does the work; pre_execute/post_execute wrap it, and on_kill handles termination.

do_x3d (deprecated) is superseded by partial() and expand(); use functools.partial or operator.partial to freeze arguments.

Trigger rules (all_success, all_failed, one_success, none_failed, etc.) are set per task via trigger_rule.

Resources (pool, pool_slots, queue) constrain where and how many task instances run at once.

Bitshift operators: task_a >> task_b (a before b), task_a << task_b (b before a), and list forms [a, b] >> c fan-in, a >> [b, c] fan-out.

set_upstream/set_downstream are the method equivalents of << and >>; set_dependency is the older alias.

cross_downstream([a, b], [c, d]) creates all-to-all edges; chain(a, b, c) and chain(*groups) linearize sequences.

TaskGroup groups tasks in the UI and namespaces task_ids as group_id.task_id; edges can cross group boundaries.

trigger_rule decides when a task runs: all_success (default), all_failed, all_done, one_success, one_failed, none_failed, none_failed_min_one_success, none_skipped, always.

A task that is skipped propagates skip downstream under all_success; use none_failed_min_one_success for join nodes that tolerate skips.

Branching: BranchPythonOperator returns the task_id(s) to follow; all non-selected branches are skipped, and the join task needs an appropriate trigger_rule (e.g., none_failed_min_one_success).

ShortCircuitOperator skips all downstream tasks when its callable returns False; it does not raise.

Dynamic mapping: task.expand(arg=[...]) creates one mapped instance per element; task.partial freezes non-mapped args; mapped tasks can be followed by a reducing task.

The scheduler creates a DAG run for each interval defined by schedule and start_date; with catchup=True it backfills every interval since start_date.

A DAG run's logical date (formerly execution_date) is the start of the interval it covers, not the time it actually ran; use data_interval_start and data_interval_end for the covered window.

For a cron schedule, data_interval_start = the cron tick, data_interval_end = the next tick; a task runs after the interval closes (unless using a custom timetable).

schedule accepts cron strings, timedelta, @daily/@hourly presets, timetables, or None (manual trigger only); schedule_interval is the legacy name.

catchup=False makes the scheduler skip missed intervals and only run the most recent; max_active_runs caps concurrent runs per DAG.

Timezones: use pendulum.timezone('UTC') or zoneinfo; DAGs run in the scheduler/airflow timezone (default UTC), and cron is evaluated in that timezone.

depends_on_past=True blocks a task instance until the same task succeeded in the previous run; wait_for_downstream extends this to downstream tasks.

end_date stops scheduling new runs; a DAG with an end_date in the past will not create new runs.

The scheduler uses a heartbeat and parses files at min_file_process_interval; the DAG's schedule is stored in serialized form, so changing it requires a parse cycle.

XComs store small values in the metadata DB keyed by (dag_id, task_id, run_id, key); xcom_push(key, value) and xcom_pull(task_ids=..., key=...) are the APIs.

A task's return value is automatically pushed under key 'return'; taskflow functions receive pulled XComs as arguments when dependencies exist.

XCom values must be JSON-serializable and small; large payloads should go to object storage and pass a URI.

Templating: fields listed in template_fields are rendered with Jinja2 using context variables (ds, ds_nodash, ts, ts_nodash, data_interval_start, data_interval_end, params, var, conn, ti, task, dag).

{{ ds }} is the logical date in YYYY-MM-DD; {{ ds_nodash }} is YYYYMMDD; {{ ts }} is ISO-8601; {{ prev_ds }} / {{ next_ds }} refer to adjacent intervals.

{{ var.value.my_var }} reads an Airflow Variable; {{ conn.my_conn.host }} reads a Connection field; {{ params.x }} reads DAG/task params.

Custom templating: add template_fields = ('my_arg',) to a custom operator; use render_template_as_native_obj=True on the DAG to keep rendered types (dict/list) native.

XComArg (task.output) is the programmatic handle for passing output between tasks and into expand(); it does not require a manual xcom_pull.

XComs are pulled at runtime, not parse time; a downstream task only sees upstream XComs after the upstream task succeeds.

Connections store host, schema, login, password, port, and extra; conn_id is the lookup key and is referenced by hooks and operators.

Hooks (BaseHook subclasses) read a Connection and expose a client (e.g., PostgresHook, S3Hook); operators typically accept a conn_id and build a hook internally.

get_connection(conn_id) resolves from the metadata DB, environment variable AIRFLOW_CONN_<CONN_ID>, or secrets backend, in that precedence order.

Variables are key-value pairs stored in the metadata DB, read via Variable.get('key') or the templated {{ var.value.key }}; JSON values can be parsed with deserialize_json=True.

Environment overrides: AIRFLOW_VAR_ sets a Variable; AIRFLOW_CONN_<CONN_ID> (URI form) sets a Connection.

Secrets backends (e.g., AWS Secrets Manager, Vault, GCP Secret Manager) are configured via secrets_backend in airflow.cfg and are queried after env vars.

Avoid calling Variable.get() at DAG parse time; it hits the DB on every parse and slows scheduling. Use templated var.value or pass via params.

Connections and Variables are encrypted at rest when a Fernet key is set; without it, secrets are stored in plaintext.

conn_id naming conventions and pools (pool, pool_slots) belong to the same operational configuration layer; pools limit concurrency per resource.

Dynamic task mapping: task.expand(x=[1,2,3]) creates one mapped task instance per element at runtime; the number of instances is determined when the upstream produces the list.

task.partial(fixed_arg=...) freezes arguments so only the expanded argument varies; combine partial and expand on the same task.

The mapped argument can be an XComArg (upstream task output), a list, or a dict; expand_kwargs maps over dicts to produce keyword arguments.

A mapped task's output is a list of results; a downstream reduce task receives it as a list (or use a normal task as the reducer).

max_mapped_tasks (DAG-level) caps the number of mapped instances; exceeding it fails the task.

Dynamic DAG generation: loop over a config list to create multiple DAGs in one file, or use a factory function returning a DAG; keep dag_id unique.

Avoid heavy work (API calls, DB queries) at module level in DAG files; it runs on every parse. Cache results in a Variable or a separate process.

TaskGroups can be mapped (expand a group) to create parallel subgraphs; the group_id must be unique per expansion.

Mapped tasks inherit the operator's template_fields and trigger_rule; each instance gets its own task_id suffix based on the map index.

retries (int) and retry_delay (timedelta) control automatic retries; retry_exponential_backoff=True increases delay exponentially, capped by max_retry_delay.

on_failure_callback, on_success_callback, on_retry_callback, and on_skipped_callback receive the context dict; callbacks run in the scheduler/worker, so keep them lightweight.

The SLA is set via sla=timedelta(...) on a task; if the task exceeds it, an sla_miss_callback fires (DAG-level) and an event is logged; SLAs do not stop the task.

Airflow 2.6+ introduced the Deadline Alerts / SLA replacement (deadline=...), which can trigger a callback and optionally mark the task failed.

exit codes: a Python exception marks the task failed; BashOperator fails on a non-zero exit code; on_kill handles SIGTERM during task termination.

AirflowException raised in a callable fails the task; AirflowSkipException marks it skipped; AirflowFailException fails immediately without retries.

emulate_default_retries and the default trigger_rule interact: a failed task with retries left stays in up_for_retry, not failed.

Email alerts: email_on_failure, email_on_retry, and email=... on the operator, or a default_args entry; requires an SMTP connection and email backend configured.

Task timeout: execution_timeout kills a running task after the limit; dagrun_timeout limits the whole DAG run.

airflow dags test <dag_id> <logical_date> runs a DAG locally without the scheduler; airflow tasks test <dag_id> <task_id> <logical_date> runs a single task.

The DagBag class loads DAG files for validation: DagBag(dag_folder=..., include_examples=False) then assert dag_id in bag.dags.

Unit tests: call the callable directly, or use task.render_template / ti.xcom_push with a mock context; pytest with airflow.utils.db.clear_db for isolation.

Common failure causes: import errors (syntax, missing provider), task_id collisions, naive start_date, and circular dependencies.

Debug UI: the Graph view shows task states; the Grid view shows runs; the task instance log is in the Logs tab and on the worker.

Performance: keep DAG files small, avoid top-level DB/API calls, reduce parse frequency via min_file_process_interval, and use pools to limit load.

Airflow 2.x uses the scheduler's serialized DAG; the parsing process runs separately, so heavy imports slow the scheduler, not the workers.

Use airflow tasks states-for-dag-run and airflow dags list-runs to inspect state from the CLI; airflow dags report summarizes DAG health.

Testing connections: airflow connections test <conn_id> verifies a connection without running a DAG.

One DAG per logical workflow; keep DAGs idempotent and deterministic so reruns produce the same result.

Use default_args for retries, owner, and email; override only where needed; keep DAG-level defaults in a shared module.

Prefer taskflow (@task) for Python logic and classic operators for integrations; avoid putting business logic in the DAG file's top level.

Set catchup=False for most new DAGs unless backfill is required; use max_active_runs to bound concurrency.

Name tasks clearly with verb-noun task_ids; group related tasks in TaskGroups; add tags and doc_md for discoverability.

Avoid XCom for large data; pass URIs or use a shared storage layer; keep variables and connections out of parse-time code.

Use pools to limit concurrency against external systems; set execution_timeout to prevent hung tasks.

Version-control DAGs, review changes, and run CI checks (DagBag import, unit tests) before deploying; keep provider versions pinned.

Delete or archive stale DAGs; unused DAGs still consume scheduler parse time and clutter the UI.