Apache Airflow 3 & Provider Packages

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.

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.

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.

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.