Before you use this reference
This page is a lookup, not a tutorial. The explanations live in Orchestration and Airflow Fundamentals, Writing Your First DAG, Operators and Tasks, Scheduling and Backfill, Task Dependencies and DAG Structure, and XCom, Connections, and Hooks.
DAG skeleton#
from datetime import datetime, timedelta
from airflow import DAGfrom airflow.operators.python import PythonOperator
from transform.raw_to_staging import load_raw_orders_to_stagingfrom marts.customer_metrics import build_customer_metrics
default_args = { "owner": "data-team", "retries": 2, "retry_delay": timedelta(minutes=5),}
with DAG( dag_id="orders_pipeline", default_args=default_args, start_date=datetime(2026, 1, 1), schedule="@hourly", catchup=False, max_active_runs=1, tags=["orders", "etl"],) as dag: load_staging = PythonOperator( task_id="load_raw_orders_to_staging", python_callable=load_raw_orders_to_staging, ) build_metrics = PythonOperator( task_id="build_customer_metrics", python_callable=build_customer_metrics, )
load_staging >> build_metricsDAG arguments#
| Argument | Meaning |
|---|---|
dag_id | Unique, stable name |
default_args | Applied to every task unless overridden |
start_date | First data interval; a fixed past date |
schedule | @hourly / cron / timedelta / None |
catchup | Run missed intervals on enable? Usually False |
max_active_runs | Concurrent runs of this DAG |
max_active_tasks | Concurrent task instances in this DAG |
tags | UI filter labels |
default_args keys#
| Key | Meaning |
|---|---|
owner | UI label |
retries | Retry count on failure |
retry_delay | timedelta between attempts |
execution_timeout | Kill the task after this timedelta |
depends_on_past | Wait for the same task in the previous run |
email_on_failure | Send on failure (needs SMTP config) |
Schedule#
schedule="@hourly" # @once @hourly @daily @weekly @monthly @yearlyschedule="30 6 * * *" # cron: 06:30 every dayschedule=timedelta(hours=6) # every 6h from start_dateschedule=None # manual trigger onlyA run for interval [T, T+1) executes shortly after T+1, not at T.
Operators#
from airflow.operators.python import PythonOperatorPythonOperator(task_id="t", python_callable=fn, op_kwargs={"k": "v"})
from airflow.operators.bash import BashOperatorBashOperator(task_id="t", bash_command="dbt run")
from airflow.operators.empty import EmptyOperatorEmptyOperator(task_id="join") # structural placeholder
from airflow.decorators import task@taskdef fn(): ...The task context (**context)#
| Key | Value |
|---|---|
data_interval_start / data_interval_end | The window this run owns |
logical_date | The interval's start (was execution_date) |
ds | logical_date as YYYY-MM-DD |
ti | Task instance — for XCom |
dag_run | The run object; dag_run.conf for manual-trigger params |
Use the interval, never datetime.now().
Dependencies#
a >> b # b after aa >> [b, c] # fan out[b, c] >> d # fan ina >> [b, c] >> d # both
from airflow.models.baseoperator import chainchain(a, b, [c, d], e)A bare [a, b, c] with no >> sets no order.
Trigger rules#
| Rule | Runs when |
|---|---|
all_success (default) | all upstream succeeded |
all_done | all upstream finished (any state) |
one_success | at least one upstream succeeded |
one_failed | at least one upstream failed |
none_failed_min_one_success | needed downstream of a branch |
XCom#
# push: return from a PythonOperator callabledef get_count(**c): return 42
# pulldef use_count(**c): n = c["ti"].xcom_pull(task_ids="get_count")
# explicit push / pull by keyc["ti"].xcom_push(key="path", value="s3://bucket/key")c["ti"].xcom_pull(task_ids="upstream", key="path")Small values only. Scoped to one DAG run.
Connections and hooks#
airflow connections add clickhouse_default \ --conn-type generic --conn-host clickhouse --conn-port 8123 \ --conn-login default --conn-password password
# or an env var (URI form):export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='generic://default:password@clickhouse:8123'from airflow.hooks.base import BaseHookconn = BaseHook.get_connection("clickhouse_default")# conn.host, conn.port, conn.login, conn.password, conn.schema, conn.extra_dejson
from airflow.providers.postgres.hooks.postgres import PostgresHookPostgresHook(postgres_conn_id="orders_postgres").get_pandas_df("SELECT ...")CLI#
airflow dags listairflow dags list-runs -d orders_pipelineairflow dags test orders_pipeline 2026-03-01 # parse + run the whole DAGairflow tasks test orders_pipeline load_staging 2026-03-01 # one task, no metadata writesairflow dags trigger orders_pipeline # manual run nowairflow dags trigger orders_pipeline --conf '{"full_refresh": true}'airflow dags backfill orders_pipeline --start-date 2026-02-01 --end-date 2026-02-28airflow dags pause orders_pipelineairflow dags unpause orders_pipelineairflow tasks list orders_pipeline --treeairflow connections list