Reference sheet · data orchestration

airflow 3

apache-airflow 3.3.0  ·  airflow.sdk 1.2.2  ·  + provider packages

A Dag file is a declaration, not a program you run. The Dag processor parses it into a serialized Dag, the scheduler decides which task instances are ready, and a worker asks the API server for work — in Airflow 3 your task code never touches the metadata database. Everything you author now comes from airflow.sdk; almost everything you integrate with comes from a provider package.

install · run · CLI Dags & scheduling tasks — the core data between tasks new in Airflow 3.x provider packages removed · gotcha ★most common

Verified July 2026 against the Airflow 3.3.0 docs, the Task SDK 1.2.2 API reference and the 3.3.0 release blog.  ·  Python ≥3.10, ≠3.15.  ·  Airflow 2 reached end of life on 22 Apr 2026 — the widely-circulated cheat sheets still teach schedule_interval, airflow db init and airflow.operators.bash; none of those work here.

Mental model — who parses your file, who decides, who runs it, and what the worker is allowed to touch
A · THE SERVICE ARCHITECTURE (AIP-72) dags/ Dag bundle Dag Processor parses → serialises Metadata DB serialised dags · runs state · XCom · connections Scheduler what is ready now? Executor Local · Celery · K8s Worker your task code API server REST /api/v2 · UI · Execution API only door to the DB airflow.sdk Task SDK 1.2.2 XComs · Variables · Connections no direct DB access from task code — Airflow 3 removed it Triggerer deferred tasks · asyncio B · WHAT YOU ACTUALLY WRITE @dag(schedule=..., catchup=False) def pipeline(): @task def extract(): ... load(transform(extract())) pipeline() calling a @task returns an XComArg, which is both the value and the edge schedule="0 3 * * *" time — a cron or timetable schedule=[Asset("s3://x")] data — an asset event schedule=None → manual @task(outlets=[asset]) a task that produces data AssetEvent emitted on success downstream Dag run no sensor, no polling Everything you integrate with lives in a provider package pip install apache-airflow-providers-amazon → from airflow.providers.amazon.operators.s3 import ... 80+ community providers · standard holds Bash/Python/File — they left airflow-core in 3.0 since May 2026 new provider releases require Airflow ≥ 3.1
↔ swipe to see the whole diagram
A complete Airflow 3 Dag — every import from airflow.sdk
# dags/etl.py  — the file is parsed by the Dag processor, not run by you
import datetime, pendulum
from airflow.sdk import dag, task, Asset          # ★ the only import path you need

orders = Asset("s3://warehouse/orders")               # a named piece of data

@dag(
    schedule="0 3 * * *",                              # bare cron → CronTriggerTimetable in 3.x
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,                                    # the default since 3.0 — state it anyway
    max_active_runs=1,
    default_args={"retries": 2, "retry_delay": datetime.timedelta(minutes=5)},
    tags=["etl"],
)
def daily_orders():
    @task
    def extract() -> list[dict]:                     # return value becomes an XCom
        return [{"id": 1, "amount": 9.99}]

    @task(outlets=[orders])                          # declares "this task produces orders"
    def load(rows: list[dict]) -> None:
        print(f"loading {len(rows)} rows")

    load(extract())                                 # the call IS the dependency edge

daily_orders()                                      # ★ you must call it, or nothing registers
Part IGetting it running install · processes · CLI · config · executors · credentials
01Install — and the package split★

airflow 3.3.0 · Python ≥3.10, ≠3.15

  • pip install "apache-airflow==3.3.0" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.3.0/constraints-3.10.txt"The constraint file is not optional. Airflow pins a large dependency tree; without it pip will happily resolve a broken environment.
  • pip install "apache-airflow[amazon,postgres,celery]==3.3.0"Extras pull in provider packages. apache-airflow is a meta-package over airflow-core, the Task SDK and the pre-installed providers.
  • pip install apache-airflow-providers-standardRead this one twice. BashOperator, PythonOperator, FileSensor and ExternalTaskSensor left airflow-core in 3.0. Installable on 2.x too, so you can migrate imports before you upgrade.moved
  • pip install "apache-airflow-task-sdk"Authoring only — lets a Dag-authoring repo or a test suite import airflow.sdk without installing the scheduler and its database drivers.
  • export AIRFLOW_HOME=~/airflowEverything (airflow.cfg, dags/, logs/, the SQLite dev DB) hangs off this.
  • docker pull apache/airflow:3.3.0Official image. For Kubernetes use the Helm chart; note every webserver key became apiServer.
02Starting the processes

what changed: three commands you must relearn

  • airflow standaloneDev only. Migrates the DB, creates an admin user and starts every component in one terminal.
  • airflow db migratedb init and db upgrade are gone; migrate both initialises and upgrades. Pair with airflow connections create-default-connections if you want the sample connections.renamed
  • airflow api-serverReplaces airflow webserver. It is no longer just a UI — it serves the React UI, the REST /api/v2, and the Execution API that workers call.renamed
  • airflow dag-processorMust now be started separately, always — including in local setups. In Airflow 2 the scheduler parsed Dag files itself.3.0
  • airflow scheduler  ·  airflow triggererThe scheduler creates runs and queues task instances; the triggerer runs the asyncio event loop that deferred tasks and asset watchers park on.
  • airflow db clean --clean-before-timestamp 2026-01-01Trim old runs, XComs and logs. Do this before a major upgrade — the 3.x schema migration is slow on a fat database.
03Two CLIs: airflow vs airflowctl

AIP-81 · apache-airflow-ctl 0.1.5

  • airflow dags list  ·  airflow dags list-import-errorslist-import-errors is the first thing to run when a Dag "doesn't appear".
  • airflow dags trigger my_dag --conf '{"env":"prod"}'Triggering now defaults to logical_date=None. --exec-date is gone.changed
  • airflow tasks test my_dag my_task 2026-07-21Runs one task, ignoring dependencies and writing no state. The fastest debug loop there is.★
  • airflow dags test my_dagExecutes a whole Dag run in-process. Prefer it over backfill when you are still developing.
  • airflowctl auth login --api-url http://localhost:8080The split: airflow is for operators who sit next to the database; airflowctl talks to a remote Airflow over the REST API, so it needs no DB credentials.3.0
  • airflow info  ·  airflow cheat-sheet  ·  airflow versioninfo dumps installed providers and their versions — the quickest way to answer "why is that import failing".
04Configuration

airflow.cfg · env vars · the migration helpers

  • AIRFLOW__CORE__PARALLELISM=64Any setting: AIRFLOW__{SECTION}__{KEY}, double underscores, uppercase. Env vars beat airflow.cfg.★
  • airflow config update --fixRewrites an Airflow 2 config to its Airflow 3 equivalents. Run airflow config lint first to see what it would touch.
  • airflow config get-value core dags_folderResolves the effective value across cfg, env and defaults — settle arguments with this, not by reading files.
  • [core] parallelism · max_active_tasks_per_dag · dag_concurrencyGlobal ceilings. Per-Dag limits (max_active_runs, max_active_tasks) sit on the Dag; pools cap a shared resource across Dags.
  • [scheduler] min_file_process_interval · parsing_processesHow often the Dag processor re-reads files. Raise the interval if parsing is burning CPU; lower it if edits feel slow to appear.
  • [core] simple_auth_manager_all_admins = TrueSimpleAuthManager is the default in 3.x. For roles and SSO, install the FAB provider and set auth_manager to FabAuthManager.3.0
05Executors

who actually launches the worker process

  • LocalExecutorSubprocesses on the scheduler host. Now works with SQLite, which is why SequentialExecutor was removed — use this for development.removed
  • CeleryExecutorDistributed workers behind Redis or RabbitMQ. Honours queue= on a task; needs the celery provider plus a result backend.
  • KubernetesExecutorOne pod per task instance. Shape it with executor_config={"KubernetesExecutor": {...}} on the task.
  • EdgeExecutorWorkers that pull work over HTTP from the API server — for remote or firewalled sites with no inbound access and no DB route.3.0
  • [core] executor = LocalExecutor,CeleryExecutorMultiple executors at once, chosen per task with executor=. This replaces CeleryKubernetesExecutor and LocalKubernetesExecutor, both removed.removed
  • # not an executor  TriggererDeferred tasks release their worker slot entirely and resume when the trigger fires — thousands of waits on one process.
06Connections, Variables & secrets★

never put a credential in a Dag file

  • airflow connections add 'my_pg' --conn-uri 'postgres://u:p@host:5432/db'A connection is a named bundle of host, login, password, schema, port and a JSON extra.
  • AIRFLOW_CONN_MY_PG='postgres://u:p@host:5432/db'Env vars define connections too, and take priority. Handy in CI, invisible to connections list.
  • from airflow.sdk import Variable, ConnectionVariable.get("k", default=None, deserialize_json=True) · Connection.get("conn_id"). Inside a task these route through the Execution API, not the DB.
  • "{{ var.value.my_key }}"  ·  "{{ conn.my_pg.host }}"In a template these are fetched lazily at render time — far better than a module-level Variable.get(), which hits the backend on every parse.gotcha
  • hook = PostgresHook(postgres_conn_id="my_pg")Hooks are the connection-consuming half of a provider. BaseHook.get_connection(id) when you need the raw object.
  • [secrets] backend = ...secrets_manager.SecretsManagerBackendVault, AWS Secrets Manager, GCP Secret Manager and friends ship as providers. Lookup order: secrets backend → env var → metadata DB.
Part IIAuthoring Dags the part you write · everything imports from airflow.sdk
07The Dag itself★

airflow.sdk.dag / airflow.sdk.DAG

  • from airflow.sdk import dag, task, DAGairflow.decorators.dag and airflow.models.dag.DAG still work but warn, and will be removed. Move now.deprecated
  • @dag(schedule="@daily", start_date=..., catchup=False) def my_pipeline(): ... my_pipeline()The trailing call is what registers the Dag. Forget it and the file parses cleanly and produces nothing.gotcha
  • with DAG(dag_id="x", schedule=None) as dag: ...The context-manager form is equally supported — use it when you build tasks in a loop or mix classic operators.
  • catchup=False # default since 3.0In Airflow 2 this defaulted to True and a Dag with an old start_date would stampede hundreds of runs on first unpause.changed
  • max_active_runs=1  ·  max_active_tasks=16Subtle: max_active_tasks was a limit on the whole Dag in Airflow 2; since 3.0 it is per run.changed
  • default_args={"retries": 2, "owner": "data-eng"}Applied to every operator as constructor kwargs. Anything set on the operator itself wins.
  • tags=["etl"] · doc_md="..." · dag_display_name · fail_fast=Truefail_fast kills running tasks on the first failure — only legal if every task uses the default trigger rule.
08@task — TaskFlow★

a Python function becomes a task

  • @task def extract() -> dict: return {"n": 1}The return value is pushed as an XCom automatically; the type hint is documentation, not enforcement.
  • @task(task_id="x", retries=3, pool="api", queue="heavy")Every BaseOperator argument is available on the decorator.
  • @task(multiple_outputs=True) def split(): return {"a": 1, "b": 2}Pushes one XCom per key, so downstream can take split()["a"]. Inferred automatically from a -> dict hint.
  • @task.bash · @task.docker · @task.kubernetes · @task.virtualenvProvider-supplied flavours, loaded dynamically from whatever you have installed.
  • @task.run_if(lambda ctx: ctx["logical_date"].day == 1)Runs only when the condition holds, else skips. @task.skip_if is the inverse. Cleaner than a branch for a simple guard.
  • from airflow.sdk import get_current_context ctx = get_current_context(); ti = ctx["ti"]Preferred over **context: it keeps your function signature honest and testable.
  • @task.stub(queue="golang") def go_step(): ...Declares a task whose implementation is a Go binary or Java jar; a coordinator runs it and proxies XComs back. Experimental.3.3
09Dependencies

four ways to draw the same edge

  • load(transform(extract()))TaskFlow: passing a result creates the edge. No >> needed — and this is the form to reach for first.★
  • a >> b >> c  ·  a >> [b, c] >> dThe classic bitshift. A list fans out and back in. Mixes freely with TaskFlow objects.
  • from airflow.sdk import chain chain(t1, [t2, t3], [t4, t5], t6)Pairs the lists element-wise: t2→t4 and t3→t5. Lists must be the same length.
  • chain_linear(op1, [op2, op3], [op4, op5], op7)Cross-connects each level to the next instead of pairing — every left node to every right node.
  • cross_downstream(from_tasks=[a, b], to_tasks=[c, d])Full mesh between two groups, without making them a chain.
  • from airflow.sdk import Label t1 >> Label("rows found") >> t2Annotates the edge in the graph view. Invaluable on branches.
  • task.output  ·  XComArg(op)Turns a classic operator into something you can pass to a TaskFlow function.
10Scheduling & timetables★

the biggest behavioural change in 3.0

  • schedule=None # the default since 3.0It was timedelta(days=1). A Dag with no schedule= now simply never runs on its own.changed
  • schedule="0 3 * * *"  ·  "@daily"  ·  timedelta(hours=6)schedule_interval and the old timetable= argument are both gone — one schedule= takes all of them.removed
  • [scheduler] create_cron_data_intervals = FalseRead carefully. A bare cron string now builds a CronTriggerTimetable, which fires at the start of the period. Airflow 2 built a CronDataIntervalTimetable, which waited a full interval. Your ds shifts.changed
  • CronTriggerTimetable("0 3 * * *", timezone="Asia/Kolkata")Pass a timetable object explicitly and you are immune to the flag above.
  • MultipleCronTriggerTimetable("0 9 * * *", "0 17 * * *", timezone="UTC")Two schedules, one Dag, at most one run per instant. EventsTimetable covers irregular calendars like holidays.
  • logical_date  ·  data_interval_start / _end  ·  run_afterlogical_date now equals run_after, not data_interval_start, and can be None. If you partition data by it, check every query.changed
  • deadline=DeadlineAlert(reference=DeadlineReference.DAGRUN_QUEUED_AT, interval=timedelta(hours=2), callback=notify)The replacement for SLAs, which were deleted outright in 3.0.3.1
11Assets — data-aware scheduling

Dataset was renamed Asset in 3.0

  • from airflow.sdk import Asset orders = Asset("s3://warehouse/orders")A URI naming a piece of data. Airflow never reads it — it is a rendezvous point, not a file handle.
  • @task(outlets=[orders]) # producerOn success the task emits an AssetEvent. That is the entire producing contract.
  • @dag(schedule=[orders, customers]) # consumerRuns when all listed assets have been updated. No sensor, no polling, no ExternalTaskSensor guesswork.★
  • schedule=AssetAny(a, b)  ·  AssetAll(a, AssetAny(b, c))Boolean composition of asset conditions, nestable.
  • AssetOrTimeSchedule(assets=[orders], timetable=CronTriggerTimetable(...))"When the data lands, or at 06:00 whichever comes first" — the usual production shape.
  • Asset("kafka://topic", watchers=[AssetWatcher(name="w", trigger=...)])Event-driven scheduling: a trigger on the triggerer materialises the asset from an external event.3.0
  • ctx["asset_state_store"].get("last_loaded_at")Assets can now carry durable state, so producer and consumer share a watermark without an external table.3.3
12XCom★

small values between tasks — never data

  • return value # pushed as key "return_value"The TaskFlow way. Explicit form: ti.xcom_push(key="k", value=v).
  • ti.xcom_pull(task_ids="extract", key="return_value")Pass a list of task_ids to gather several at once.
  • "{{ ti.xcom_pull(task_ids='extract') }}"In a template field, for classic operators.
  • # XCom pickling was REMOVED in 3.0enable_xcom_pickling is gone on security grounds. Values must be JSON-serialisable, or you write a custom XCom backend.removed
  • [core] xcom_backend = mypkg.S3XComBackendThe escape hatch for large or non-JSON payloads: store the object elsewhere, keep a reference in the DB.
  • # rule of thumbXCom rows live in the metadata database. Pass identifiers and counts, never DataFrames. If it wouldn't fit comfortably in a log line, write it to object storage and pass the path.gotcha
  • ctx["task_state_store"].set("last_cursor", n)Durable per-task state that survives retries and reruns — what people used to abuse XComs and Variables for.3.3
13Params & templating

Jinja renders at run time, not parse time

  • params={"env": Param("dev", enum=["dev", "prod"])}Typed, validated run parameters. The UI builds a trigger form from the schema.
  • "{{ params.env }}"  ·  "{{ dag_run.conf['env'] }}"params is validated and has defaults; dag_run.conf is the raw untyped payload.
  • "{{ ds }}" · "{{ ds_nodash }}" · "{{ ts }}" · "{{ logical_date }}"The survivors. Removed in 3.0: execution_date, next_ds, prev_ds, tomorrow_ds, yesterday_ds and their _nodash twins — these now raise.removed
  • data_interval_start · data_interval_end · prev_end_date_successThe correct way to express "since the last successful run".
  • template_searchpath=["/opt/sql"]; sql="q.sql"Operators declare template_ext, so a .sql or .sh filename is loaded and rendered.
  • from airflow.sdk import literal bash_command=literal("echo {{ not_jinja }}")Opts a template field out of rendering — the fix for shell or SQL that legitimately contains braces.
  • render_template_as_native_obj=TrueRenders to real Python types instead of strings, so "{{ params.n }}" arrives as an int.
14Dynamic task mapping

a task count decided at run time

  • process.expand(x=[1, 2, 3])Three mapped instances, indices 0..2. The list can be an XCom from an upstream task — that is the whole point.★
  • process.expand(x=get_files())The fan-out width is not known until get_files finishes; the scheduler expands afterwards.
  • process.partial(conn_id="pg").expand(sql=queries)partial() pins the constant arguments; expand() varies the rest.
  • op.expand_kwargs([{"a": 1, "b": 2}, {"a": 3, "b": 4}])Vary several arguments together rather than taking a cross product.
  • a.expand(x=[1,2], y=[3,4]) # → 4 instancesMultiple expand keys multiply. Easy to detonate accidentally — cap with max_active_tis_per_dag.gotcha
  • map_index_template="{{ task.op_kwargs['name'] }}"Labels the mapped instances in the UI with something meaningful instead of a bare integer.
  • ti.xcom_pull(task_ids="process") # → list of all map resultsThe reduce half: a downstream task receives every mapped return value as a list.
15Grouping, setup & teardown

structure without SubDAGs

  • @task_group(group_id="ingest") def ingest(src): ...Collapsible in the UI, and it takes and returns values like any TaskFlow function.
  • with TaskGroup("ingest", prefix_group_id=False) as tg:By default child ids become ingest.extract. Turn the prefix off if downstream code refers to bare ids.
  • ingest.expand(src=sources)Whole task groups can be dynamically mapped, not just single tasks.
  • # SubDAGs were REMOVED in 3.0Refactor to task groups (visual nesting) or assets (cross-Dag dependency). The old SubDagOperator is not coming back.removed
  • @setup def make_cluster(): ...  ·  @teardown def kill_cluster(): ...Teardown runs even when the work fails, and its own failure does not fail the run unless you say so.
  • @teardown(on_failure_fail_dagrun=True)Use when a leaked resource is itself an incident worth failing on.
  • work.as_teardown(setups=make_cluster)Binds the pair explicitly so clearing the work also re-runs the setup.
16Branching & trigger rules

where skips propagate and surprise you

  • @task.branch def pick(): return "path_a"Return the task_id (or a list) to run; every other direct downstream is skipped.
  • @task.short_circuit def has_rows(): return n > 0Falsy return skips everything downstream. The cleanest "nothing to do today" guard.
  • trigger_rule="all_success" # the defaultAnd this is why a join after a branch mysteriously skips: a skipped parent is not a success.gotcha
  • trigger_rule="none_failed_min_one_success"The join rule. Runs if nothing failed and at least one parent actually ran — what you almost always want after a branch.★
  • "all_done" · "one_failed" · "one_success" · "all_skipped" · "always"all_done for cleanup that must run either way; one_failed for an alerting task.
  • raise AirflowSkipException  ·  raise AirflowFailExceptionSkip deliberately, or fail without consuming retries when the failure is permanent.
  • depends_on_past=True  ·  wait_for_downstream=TrueSerialises a task against its own history. Powerful and a classic way to deadlock a backfill.gotcha
17Sensors & deferrable operators

waiting without burning a worker slot

  • FileSensor(task_id="wait", filepath="/data/in.csv", poke_interval=60)Defaults: poke_interval=60, mode="poke". Note it now lives in the standard provider.
  • mode="reschedule"Frees the worker slot between pokes. Use it whenever the wait may exceed a few minutes; keep poke_interval above 60s so you don't hammer the scheduler.★
  • deferrable=TrueBetter still: the task parks on the triggerer and consumes no slot at all. Most cloud provider operators accept this flag.
  • soft_fail=True  ·  never_fail=TrueSkip instead of fail on timeout / on any exception. Mutually exclusive — setting both is an error.
  • exponential_backoff=True, max_wait=timedelta(minutes=10)Progressively longer gaps between pokes, capped.
  • timeout=3600 # vs execution_timeouttimeout counts from the first poke including reschedule gaps; execution_timeout counts only running time. In reschedule mode they differ wildly.gotcha
  • PokeReturnValue(is_done=True, xcom_value=payload)Lets @task.sensor return what it found rather than merely announcing that it found it.
Part IIIProvider packages 80+ community packages · where every integration actually lives
18How a provider works★

the naming rule is mechanical

  • pip install apache-airflow-providers-amazon from airflow.providers.amazon.operators.s3 import S3CreateObjectOperatorDistribution name → import path: swap the hyphens for dots and drop the prefix. Learn this once and you never search for an import again.
  • airflow providers list  ·  airflow providers get amazonWhat is installed, at what version, and what it registers.
  • # the four things inside a providerHooks wrap a connection · Operators do one unit of work · Sensors wait · Transfers move X→Y. Providers may also add connection types, secrets backends, log handlers, notifiers and executors.
  • pip install "apache-airflow[amazon]==3.3.0"The extra installs the matching provider at a version the constraint file has tested together.
  • # versioningProviders release on their own cadence with their own semver, independent of Airflow core. Pinning core does not pin them.gotcha
  • # since May 2026Newly released providers require Airflow ≥ 3.1. Provider-supplied CLI commands need core ≥ 3.2.2026
19standard — the one you must install★

apache-airflow-providers-standard

  • from airflow.providers.standard.operators.bash import BashOperatorNot airflow.operators.bash. This single move breaks more Airflow 2 Dags than anything else in the upgrade.moved
  • from airflow.providers.standard.operators.python import PythonOperator, BranchPythonOperator, PythonVirtualenvOperatorThough with TaskFlow you rarely instantiate PythonOperator by hand any more.
  • from airflow.providers.standard.sensors.external_task import ExternalTaskSensorAlso sensors.filesystem.FileSensor, sensors.date_time, sensors.time_delta.
  • EmptyOperator(task_id="join")The renamed DummyOperator — a structural placeholder that does nothing.
  • TriggerDagRunOperator(trigger_dag_id="other", wait_for_completion=True)Explicit cross-Dag triggering. Prefer assets when the real relationship is "when this data is ready".
  • LatestOnlyOperator  ·  ShortCircuitOperatorLatestOnly skips downstream on backfilled runs — useful for anything that publishes to a live target.
20Cloud providers

amazon · google · microsoft.azure

  • from airflow.providers.amazon.aws.operators.s3 import S3CreateObjectOperatorNote the extra aws segment — the amazon provider nests everything under it. Hooks: ...aws.hooks.s3.S3Hook.
  • S3KeySensor(bucket_key="s3://b/k", deferrable=True)Also GlueJobOperator, EmrServerlessStartJobOperator, LambdaInvokeFunctionOperator, RedshiftDataOperator, EcsRunTaskOperator.
  • from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperatorThe one to know for GCP. Also GCSToBigQueryOperator, DataprocSubmitJobOperator, CloudRunExecuteJobOperator.
  • from airflow.providers.microsoft.azure.operators.wasb_delete_blob import ...Plus AzureDataFactoryRunPipelineOperator and the Synapse and Container Instances operators.
  • aws_conn_id="aws_default" # or None to use the instance rolePassing None falls back to the ambient credential chain — the right answer on EKS or EC2.
  • cncf.kubernetes → KubernetesPodOperatorThe universal escape hatch: run any container as a task, regardless of your executor.★
21Databases & SQL

common.sql is the shared base

  • from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperatorThe modern way. PostgresOperator, MySqlOperator, SnowflakeOperator and the rest are deprecated in favour of this one operator plus a conn_id.deprecated
  • SQLExecuteQueryOperator(conn_id="pg", sql="queries/daily.sql")A .sql path is loaded and Jinja-rendered because sql is a template field with template_ext.
  • SQLColumnCheckOperator  ·  SQLTableCheckOperatorDeclarative data-quality gates — null rates, uniqueness, row-count bounds — without writing assertion code.
  • hook = PostgresHook(postgres_conn_id="pg") rows = hook.get_records(sql)Also get_pandas_df(), insert_rows(), bulk_load(). All DbApiHook subclasses share this surface.
  • postgres · mysql · snowflake · databricks · trino · mongo · elasticsearchEach is its own distribution; install only what you use, or your image grows without limit.
  • SQLThresholdCheckOperator(min_threshold=..., max_threshold=...)Fails the run when a metric leaves its band — a cheap regression alarm on top of an existing query.
22Moving data & object storage

transfers, and the path abstraction

  • S3ToRedshiftOperator · GCSToBigQueryOperator · SFTPToS3OperatorTransfer operators are named XToYOperator and live in the provider of one of the two ends — usually the destination.
  • from airflow.sdk import ObjectStoragePath p = ObjectStoragePath("s3://bucket/key", conn_id="aws_default")A pathlib-style API over S3, GCS and Azure. p.read_text(), p.iterdir(), p.copy(dst) — and the target can be a different cloud.★
  • p.write_bytes(data)  ·  with p.open("rb") as f:Backed by fsspec, so it streams rather than buffering whole objects.
  • http · ftp · ssh · sftp · smtp · slackThe small utility providers. HttpOperator plus HttpSensor covers most REST integrations without a custom operator.
  • docker · apache.spark · apache.kafka · dbt.cloudDbtCloudRunJobOperator with wait_for_completion and deferral is the standard Airflow↔dbt seam.
  • # pick the ends carefullyA transfer operator pulls bytes through the worker. For large volumes, prefer an operator that tells the warehouse to do the load itself.gotcha
23Newer & infrastructure providers

the 2025–26 additions

  • pip install apache-airflow-providers-common-aiLLM and agent operators built on pydantic-ai: 6 operators, 5 toolsets, 20+ model providers behind one connection type.2026
  • apache-airflow-providers-fabRequired if you want roles, permissions or SSO — Flask-AppBuilder left core when the UI became React, and SimpleAuthManager took its place.moved
  • apache-airflow-providers-gitA Dag bundle backend that pulls Dags straight from a repository at a pinned revision — the mechanism behind Dag versioning.3.0
  • apache-airflow-providers-openlineageEmits column-level lineage automatically from supported operators. Mostly configuration rather than Dag code.
  • apache-airflow-providers-edge3Ships the EdgeExecutor and its worker: remote sites poll the API server over HTTPS, no inbound ports, no database route.
  • apache-airflow-providers-celery · -cncf-kubernetesEven the mainstream executors are providers now — core ships only LocalExecutor.
Part IVOperating, migrating, and the traps testing · observability · the API · the 2→3 checklist
24Migrating Airflow 2 → 3★

Airflow 2 went EOL on 22 Apr 2026

  • ruff check dags/ --select AIR301 --fixStart here. AIR301/302 flag genuinely breaking changes; AIR311/312 flag recommended ones. Needs ruff ≥ 0.13.1. Add --unsafe-fixes for the import-path rewrites.
  • airflow config lint  →  airflow config update --fixDoes the same job for airflow.cfg.
  • # you must be on 2.7+ to upgrade at allGo to the latest 2.x first so you see every deprecation warning, then jump. There is no path from 2.6 or earlier.
  • airflow.datasets.Dataset → airflow.sdk.AssetAlong with DatasetAlias/All/Any → AssetAlias/All/Any, and every airflow.decorators.* and airflow.models.* authoring import → airflow.sdk.
  • schedule_interval= → schedule=Plus timetable= → schedule=, and audit anything reading execution_date.
  • # invalid kwargs now fail at importallow_illegal_arguments is gone, so a typo like num_partition= that Airflow 2 silently ignored raises TypeError and breaks the whole Dag file.changed
25Testing Dags

catch it before the scheduler does

  • python dags/my_dag.pyThe cheapest check there is: if the file raises, the Dag processor will too.
  • airflow tasks test my_dag my_task 2026-07-21One task, real code, no state written and no dependency checks.★
  • airflow dags test my_dagA full run in one process — the closest thing to an integration test.
  • dag.test() # from pytestSame execution path, callable from a test so you can assert on the outcome.
  • def test_no_import_errors(): assert DagBag().import_errors == {}The one test every repository should have. Add assertions for required tags, owners and retry counts while you are there.
  • extract.function() # call the undecorated callable@task keeps the original function available, so the business logic can be unit-tested with no Airflow at all.
26Logging & observability

where the output actually goes

  • import logging; log = logging.getLogger(__name__)Anything logged or printed inside a task is captured and shown in the task-log view.
  • [logging] remote_logging = True remote_base_log_folder = s3://bucket/airflow-logsEssential for Kubernetes and Celery, where the pod holding the log is gone by the time you look.
  • mask_secret(token)From airflow.sdk.log. Connection passwords are masked automatically; anything you fetch yourself is not.
  • [metrics] otel_on = True · statsd_on = TrueOpenTelemetry metrics and traces. 3.3 breaking change: timers are emitted as Histograms rather than Gauges, so existing dashboards need requerying.3.3
  • on_failure_callback=slack_notifierSet it in default_args to cover every task. Notifiers (BaseNotifier subclasses) ship with the slack, smtp and pagerduty providers.
  • dag_processing.last_run.seconds_agoNow emitted with file_path/bundle_name tags instead of the filename baked into the metric path — another dashboard-breaking change in 3.3.3.3
27The REST API & clients

/api/v2, FastAPI-based

  • POST /api/v2/dags/{dag_id}/dagRuns/api/v1 is removed. Get a token from /auth/token and send it as a bearer.removed
  • pip install apache-airflow-clientThe official Python client — and the supported replacement for code that used to query the metadata DB directly.
  • GET /api/v2/dags/{id}/dagRuns/{run_id}/waitBlocks until the run finishes and streams NDJSON, returning the value of the task marked @result. Lets a Dag sit behind an API endpoint. Experimental.3.3
  • airflowctl dags trigger my_dagThe same API, wrapped in a CLI, from anywhere with network access and a token.
  • # the auth route movedAuth manager routes are prefixed with /auth, so an OAuth redirect of /oauth-authorized/google is now /auth/oauth-authorized/google.changed
  • # do not query the metadata DBThe schema is explicitly not a public API. The DbApiHook workaround is documented only as a last resort and is expected to break.gotcha
28Practices that keep it fast

the Dag file is parsed over and over

  • # top level of a Dag file = hot pathIt re-executes on every parse cycle. No API calls, no queries, no Variable.get(), no heavy imports at module scope.★
  • start_date=pendulum.datetime(2026, 1, 1, tz="UTC")Always static and timezone-aware. A dynamic start_date such as days_ago(1) moves under you and misbehaves.gotcha
  • # tasks must be idempotentRetries, clears and backfills all re-run the same task instance. Write DELETE ... WHERE ds = ... then insert, rather than a bare append.
  • pool="warehouse", pool_slots=2The right way to protect a fragile downstream system, since it caps concurrency across every Dag at once.
  • execution_timeout=timedelta(hours=1)Without it a hung task holds its slot indefinitely. Set it on anything that talks to a network.
  • retry_exponential_backoff=2.0, max_retry_delay=timedelta(hours=1)Changed: this is now a float multiplier (0 disables, 2.0 doubles each time), not the boolean it was in Airflow 2.changed
29Failure modes worth recognising

symptom → cause

  • # "my Dag doesn't show up"In order: you forgot to call the @dag function; airflow dags list-import-errors; the file is outside the bundle; the Dag processor isn't running.
  • # "it's scheduled but nothing runs"It is paused; or catchup=False plus a past start_date means nothing is due yet; or max_active_runs is already saturated.
  • DuplicateTaskIdFoundTwo tasks share a task_id — usually a loop that forgot to interpolate the index into the id.
  • # the join after my branch is skippedSkips propagate under all_success. Use none_failed_min_one_success on the join.gotcha
  • TypeError: Invalid arguments were passedAn operator kwarg that Airflow 2 ignored. Since 3.0 it fails the import of the entire file, not just that task.
  • AirflowTaskTimeout # vs AirflowSensorTimeoutThe first is execution_timeout (running time), the second is a sensor's timeout (wall time since the first poke). In reschedule mode these diverge enormously.
1 · Where your code actually runsfour processes, three of them execute something you wrote
YOUR DAG FILE CONTAINS TWO KINDS OF CODE top level — the declaration @dag(...) · @task(...) · imports · loops re-runs on EVERY parse cycle runs on: Dag Processor function bodies — the work everything inside a @task runs once per task instance runs on: Worker AND EACH PROCESS SEES A DIFFERENT WORLD Dag Processor parses your files DB: yes your top-level code Scheduler reads serialised dags DB: yes none of your code Triggerer async waits DB: yes your trigger code Worker runs task instances DB: NO your task bodies API server only so the two classic mistakes are: 1. an API call at the top level — it fires every parse cycle, not once per run 2. a session query inside a task — worked in Airflow 2, impossible in Airflow 3
↔ swipe to see the whole diagram
2 · The cron default that movedwhy your ds shifts by one interval on upgrade
the same Dag: schedule="0 0 * * *" Airflow 2 — CronDataIntervalTimetable the cron describes the DATA INTERVAL, so the run waits for the interval to close Mon 00:00 Tue 00:00 Wed 00:00 data interval run fires here logical_date = Mon 00:00 = data_interval_start — a full day BEHIND the clock Airflow 3 — CronTriggerTimetable (the new default) the cron describes WHEN TO RUN, like POSIX cron — it fires immediately Mon 00:00 Tue 00:00 Wed 00:00 logical_date = run_after = Mon 00:00 — and it may be None for manual runs keep the old behaviour: pass a CronDataIntervalTimetable, or set create_cron_data_intervals = True
↔ swipe to see the whole diagram
3 · Reading a provider namepackage name and import path are the same string
WHAT YOU PIP INSTALL apache-airflow-providers-amazon ← the only part that varies WHAT YOU IMPORT airflow.providers.amazon.aws.operators.s3 some providers nest (google.cloud, microsoft.azure) operators | hooks sensors | transfers the service THE ONE THAT CATCHES EVERYONE Airflow 2 — in core airflow.operators.bash airflow.operators.python airflow.sensors.filesystem every online cheat sheet still says this Airflow 3 — standard provider airflow.providers.standard.operators.bash airflow.providers.standard.operators.python airflow.providers.standard.sensors.filesystem installable on 2.x too — migrate before you upgrade
↔ swipe to see the whole diagram
4 · Four ways to draw one graphthe same fan-out, written four times
THE GRAPH t1 t2 t3 t4 t1 fans out to t2 and t3, both of which feed t4. t4 needs trigger_rule="all_success" to be safe only if t2/t3 cannot skip 1 · TaskFlow — prefer this a = t1() t4(t2(a), t3(a)) the edge and the value are the same thing — no way for them to disagree 2 · bitshift t1 >> [t2, t3] >> t4 shortest to read; the only option when tasks pass nothing to each other 3 · chain chain(t1, [t2, t3], t4) careful: chain(a,[b,c],[d,e]) PAIRS them b→d and c→e, and the lists must match chain_linear cross-connects instead 4 · cross_downstream cross_downstream([t1], [t2, t3]) cross_downstream([t2, t3], [t4]) full mesh between two groups without making them a chain
↔ swipe to see the whole diagram
Worth memorising
from airflow.sdk import ...
The only authoring import path in Airflow 3. airflow.decorators, airflow.models, airflow.datasets all still work but warn, and are scheduled for removal.
my_pipeline()
A @dag function must be called at module level. Miss it and the file parses perfectly and registers nothing.
Dataset → Asset
Renamed in 3.0, along with DatasetAlias/All/Any. The concept is unchanged: a URI that producers declare as an outlet and consumers put in schedule=.
schedule= only
schedule_interval= and timetable= are both gone. One argument takes a cron string, a timedelta, a timetable or a list of assets.
catchup defaults to False
Reversed in 3.0. The old default was the reason an unpaused Dag with a year-old start_date would launch hundreds of runs at once.
bare cron → CronTriggerTimetable
It fires at the start of the period rather than waiting for a data interval to close, so logical_date and every ds derived from it shift by one interval on upgrade.
logical_date == run_after
No longer equal to data_interval_start, and it can be None. execution_date is deleted, as are next_ds, prev_ds, tomorrow_ds and yesterday_ds.
BashOperator moved
It, PythonOperator, FileSensor and ExternalTaskSensor live in apache-airflow-providers-standard now. Install it or nothing imports.
no DB access from tasks
Task code talks to the API server through the Task SDK. Session queries and model imports that worked in Airflow 2 now fail — use the Airflow Python Client.
top level runs constantly
Everything outside a task body re-executes on every parse cycle. A Variable.get() or an API call up there is a request every few seconds, forever.
none_failed_min_one_success
The trigger rule for a join after a branch. Under the default all_success, a skipped parent skips the join, which is the single most common "why did nothing run" report.
db migrate, api-server, dag-processor
db init/db upgrade are gone, webserver is renamed, and the Dag processor is now a separate process you must start yourself even locally.
retry_exponential_backoff is a float
A multiplier now: 0 disables, 2.0 doubles each retry. It was a boolean in Airflow 2, so an unchanged True no longer means what you think.
invalid kwargs break the file
allow_illegal_arguments is gone. A misspelled operator argument that Airflow 2 ignored silently now raises TypeError at import and takes the whole Dag file down.