Airflow Theoretical

The reasoning

Directed means each dependency has a direction: extract runs before load. Acyclic means no cycle: load cannot also be upstream of extract. Airflow rejects a cycle at parse time, because a cycle would never finish. The graph is the whole contract between tasks.

The file is ordinary Python, parsed by the dag processor on a loop. It instantiates `DAG(dag_id='orders_daily', schedule='@daily', start_date=datetime(2026, 1, 1))`, creates tasks from operators or `@task` functions, and wires them: `extract >> transform >> load`. Anything at module level runs every parse, so the file must be cheap.

A DAG run is one execution of the DAG for one logical date. A task instance is one task in one run. The UI, the metadata database and every log line are keyed by those three: dag_id, task_id, run.

Airflow 3, released in 2025, keeps this model and changes the machinery around it. It adds a separate API server, a dag processor, a task execution API, and `logical_date` as the one date name. The Python-as-configuration idea is what has not changed since 2015.

The reasoning

The dag processor reads the DAG folder on a loop, executes each file, and writes the serialised DAG to the database. In Airflow 2 this lived inside the scheduler; Airflow 3 runs it as its own process so parsing cannot stall scheduling.

The scheduler reads serialised DAGs, decides which runs are due from each timetable, creates task instances, checks dependencies and pools, and hands runnable tasks to the executor. It is the heart of the system and can run with several replicas.

The executor is a component inside the scheduler process that places tasks. LocalExecutor runs them as subprocesses on the scheduler host, CeleryExecutor sends them to a queue for worker machines, and KubernetesExecutor launches one pod per task. Airflow 2.10 added running several executors at once.

The API server replaced the webserver. Besides the UI and the public REST API, it serves the task execution API that workers call for connections, variables and XComs. Tasks no longer open a database connection themselves. The triggerer runs the asyncio loop that deferrable operators wait on. The metadata database, PostgreSQL in production, holds all state.

The reasoning

An operator is a class that defines one unit of work. Instantiating it in a DAG with a task_id makes a task. Action operators call an external system or run code; they are the majority. The `@task` decorator is a PythonOperator written as a function.

Transfer operators are a convention: a hook for the source, a hook for the destination, and a copy between them, named Source To Destination. They are convenient for small data and a trap for large data, because the bytes pass through the worker.

Sensors are operators whose `poke` method returns true or false. They subclass BaseSensorOperator and gain `poke_interval`, `timeout`, `mode` and `soft_fail`. Deferrable versions hand the wait to the triggerer instead of a worker.

Airflow 3 moved BashOperator, PythonOperator, EmptyOperator and the common sensors out of core into `apache-airflow-providers-standard`, which is installed by default. Every cloud or database operator is in its own provider, versioned separately, which is why an import path starts with `airflow.providers`.

The reasoning

The sensor's `poke(context)` method is called repeatedly. In poke mode the task process stays alive and occupies one of the worker's slots the whole time. Ten sensors waiting an hour each is ten slots doing nothing, which is how a pool fills with waiting and real work queues behind it.

`mode='reschedule'` ends the task process after each failed check and marks the instance up_for_reschedule. The scheduler starts it again after poke_interval. The slot is free in between. The cost is scheduler work and a start-up per check, so the interval should be a minute or more.

`timeout` is the total wait in seconds before the sensor fails, 60 times 60 times 24 times 7 by default. `soft_fail=True` marks it skipped instead of failed on timeout. `exponential_backoff` lengthens the interval over time. `poke_interval` and `timeout` should be set explicitly on every sensor.

Deferrable sensors, `deferrable=True` on many provider sensors, suspend the task entirely and hand an async trigger to the triggerer, which waits on thousands of conditions in one process. They are the right answer for long waits on systems that support it, and reschedule is the fallback.

The reasoning

A hook takes a `conn_id`, fetches the connection from the secrets chain, and exposes methods: `PostgresHook(postgres_conn_id='dwh').get_conn()` returns a DB-API connection; `S3Hook().load_file(...)` uploads; `SnowflakeHook().run(sql)` executes. Credentials never appear in the DAG.

Every operator is a thin wrapper around a hook. `SQLExecuteQueryOperator` builds a hook from its conn_id and calls `run`. That is why a custom operator is usually twenty lines: construct the hook in `execute`, call it, return a value.

Use a hook directly when the logic is more than one call: fetch from an API, transform in pandas, write to S3. A `@task` function that creates hooks is cleaner than three operators passing data through XCom.

Hooks live in provider packages next to their operators and define the connection type's form fields in the UI. Hooks must be instantiated inside a task, not at module level, because module level runs on every parse and would open connections on the dag processor.

The reasoning

Before Airflow 2 everything lived in one package and a bug fix to the S3 hook waited for an Airflow release. Since 2.0 each integration is its own package with its own changelog, released on its own cadence, often monthly.

Installing `apache-airflow-providers-amazon` adds `airflow.providers.amazon.aws.hooks.s3.S3Hook`, the S3 connection type with its form, dozens of operators and sensors, and in recent versions an AWS executor and a secrets backend. The package declares which Airflow versions it supports.

Independent versioning means an upgrade is two decisions: Airflow core and each provider. A constraints file published per Airflow release pins compatible versions, and the recommended install uses it. Mixing a new provider with an old core fails at import.

Airflow 3 extracted the remaining core operators into `apache-airflow-providers-standard`: BashOperator, PythonOperator, EmptyOperator, the branch operators and the basic sensors. It is installed with Airflow, but the import path changed, which is the first migration edit most DAGs need.

The reasoning

The classic style is `PythonOperator(task_id='extract', python_callable=extract)` plus `extract >> transform` plus `ti.xcom_pull(task_ids='extract')`. TaskFlow collapses it: `@task def extract(): return rows`, `@task def transform(rows): ...`, and `transform(extract())`. The call graph is the dependency graph.

Return values go to XCom under the key `return_value`, serialised as JSON by default. The size limits of XCom apply, so a function must return a pointer such as an S3 path, not a DataFrame. `multiple_outputs=True` on a function returning a dict pushes each key as its own XCom.

`@task.virtualenv` runs the function in a fresh virtual environment with its own requirements, isolating dependencies. `@task.kubernetes` runs it in a pod. `@task.bash`, from 2.9, returns the command string to run. `@task.branch` returns the task ids to follow. `@task.sensor` builds a sensor from a function returning PokeReturnValue.

TaskFlow and classic operators mix freely. A decorated function can take an operator's `.output` as an argument, and an operator can receive a TaskFlow return value as a templated parameter. In Airflow 3 the decorators come from the Task SDK, `airflow.sdk`.

The reasoning

Before 2.3, running one task per file meant either knowing the files at parse time or generating the DAG dynamically. Mapping moves the decision to run time: a `list_files` task returns the paths, and `process.expand(path=list_files())` creates one `process` instance per path.

`partial` holds the arguments that are the same for every copy; `expand` holds the ones that vary. `expand_kwargs` takes a list of dicts to vary several arguments together. Classic operators map too: `BashOperator.partial(task_id='run').expand(bash_command=cmds)`.

Each mapped instance has a map index, shows as its own row in the grid view, and retries independently. A downstream task that takes the mapped task's output receives the list of all results, which is the reduce step. Mapping over a mapped output chains fan-outs.

`max_map_length`, 1024 by default in `[core]`, caps the fan-out and fails the task that tried to exceed it. An empty list produces zero instances and the task is marked skipped. Mapped tasks respect pools and `max_active_tis_per_dag`, so a thousand-way fan-out does not run a thousand at once.

The reasoning

XCom is a row keyed by dag_id, task_id, run_id, map_index and key. The default backend writes the JSON into that row. The database limits apply, a blob column on MySQL, a large text on PostgreSQL, and a table full of multi-megabyte values makes the scheduler and UI slow.

A custom backend subclasses BaseXCom. `serialize_value` can check the size, upload anything over a threshold to S3 or GCS, and return the path; `deserialize_value` recognises the path and downloads. Tasks are unchanged; `xcom_pull` still returns the object.

`apache-airflow-providers-common-io` provides `XComObjectStorageBackend`. Configure `[common.io] xcom_objectstorage_path` with a bucket URL and `xcom_objectstorage_threshold` with the byte size above which values go to storage; below it they stay in the database. The default threshold of minus one keeps everything in the database until you set it.

The design rule survives the backend: pass pointers, not data. The backend is a safety net for medium-sized values, not a reason to ship DataFrames through XCom. In Airflow 3, XCom reads and writes go through the task execution API, and the backend runs on the API server side.

The reasoning

`default_args={'owner': 'data-eng', 'retries': 2, 'retry_delay': timedelta(minutes=5)}` means every operator in the DAG gets those three unless it sets its own. `BashOperator(task_id='x', retries=0)` wins for that task. The dict is merged into the operator's constructor arguments.

Operators accept only the arguments BaseOperator and the specific class define. Putting a key no operator accepts in default_args raises at parse time in recent versions, which catches typos such as `retry` for `retries`.

Retries default to 0 unless `[core] default_task_retries` is set, so a DAG without default_args has no retries. That is the first thing to put in. `on_failure_callback` and `sla` historically went here too; SLAs were removed in Airflow 3.

start_date in default_args is a legacy pattern from before DAGs had their own start_date argument. It still works per task but is confusing, because the DAG's start_date is what the scheduler uses. Set it on the DAG.

The reasoning

`DAG(..., params={'target_date': Param('2026-01-01', type='string', format='date'), 'dry_run': Param(False, type='boolean')})`. The UI renders a date picker and a checkbox when you press trigger, and rejects a value of the wrong type.

A scheduled run uses the defaults. A manual run, from the UI or `POST /dags/{id}/dagRuns` with a `conf` body, merges the supplied values over the defaults. Since 2.6, `conf` values that violate the Param schema are rejected, which stopped a class of typo bugs.

In a template, `{{ params.target_date }}` renders the value. In a `@task` function, `context['params']['target_date']` or a `params` keyword argument reads it. `render_template_as_native_obj=True` keeps the Python type instead of a string.

Params replace the older pattern of reading `dag_run.conf.get('x')` directly, which had no defaults and no validation and broke scheduled runs where conf is empty. Task-level params override DAG-level ones for that task only.

The reasoning

Before 2.7 the stop-cluster task needed `trigger_rule='all_done'` and still confused the DAG's final state, because a failed job followed by a successful stop looked like success. Setup and teardown make the intent explicit and fix the run state logic.

`create_cluster.as_setup()`, `delete_cluster.as_teardown(setups=create_cluster)`, and `create_cluster >> run_job >> delete_cluster`. Or `with create_cluster >> delete_cluster: run_job()`, which wires everything inside the block as work. Setups and teardowns show with distinct icons in the grid.

Semantics: the teardown runs once its setup succeeded, regardless of the work tasks' outcome, so resources are always released. The DAG run's state is decided by the work tasks, not the teardown. Clearing a work task also clears its setup and teardown so the resource is recreated for the rerun.

`on_failure_fail_dagrun=True` on the teardown makes a failed cleanup fail the run, for cases where a leaked cluster is itself an incident. Teardowns can also have their own setups, and TaskGroups can have setup and teardown at the group level.

The reasoning

`with TaskGroup('load') as load: extract >> transform`, or `@task_group def load(): ...`. Tasks inside get ids like `load.extract` unless `prefix_group_id=False`. Groups nest, can be mapped with expand, and can have setup and teardown. Dependencies set on the group apply to its first and last tasks.

The SubDagOperator created a whole DAG run for the child inside one task of the parent. The child used the SequentialExecutor by default, so its tasks ran one at a time, and it occupied a parent slot while waiting for child slots, which deadlocked pools. Its state and the parent's disagreed in the UI.

TaskGroups solved the original need, organising a large DAG, without a second scheduling layer. Everything in the group is a normal task instance of the parent DAG, visible in the grid, retrying and pooling like any other.

What a TaskGroup does not provide is isolation or reuse across DAGs. For a reusable unit across DAGs, the options are a shared Python function that builds the group, a separate DAG triggered with TriggerDagRunOperator, or an asset dependency.

The reasoning

With all_success, a skipped upstream propagates: the task is skipped, and so is everything below it. That is correct for a straight pipeline and wrong after branching, where one path is always skipped by design.

`none_failed_min_one_success` runs when no upstream failed and at least one succeeded, which is exactly the join condition. `none_failed` is similar but also runs when all upstreams were skipped. `all_done` runs regardless of outcome, the classic cleanup rule before setup and teardown existed.

`one_success` and `one_failed` fire as soon as one upstream reaches that state, without waiting for the rest, useful for an alert task. `all_failed` runs only when every upstream failed. `always` ignores upstream entirely and is rarely right.

The rule is set per task, `trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS`, and interacts with `depends_on_past`, which adds a condition on the previous run of the same task. The grid view shows a skipped join in pink, which is how the bug is usually found.

The reasoning

`@task.branch def choose(**context): return 'full_load' if context['params']['full'] else 'incremental_load'`. The returned id must be a direct downstream of the branch task. Returning a list runs several paths; returning None skips all of them.

Skipped is a terminal state with no retry, and the unchosen path's tasks never run. The downstream join must use `none_failed_min_one_success` or the skip swallows it. Branching therefore shapes the trigger rules of everything after it.

`ShortCircuitOperator` and `@task.short_circuit` are the one-path version: return true to continue, false to skip all downstream. `ignore_downstream_trigger_rules=False` makes it respect downstream rules instead of forcing the skip.

Branch operators return at run time, so the decision can use data: a row count, a file's presence, a param. The alternative, deciding at parse time with an `if` in the DAG file, produces a different graph per parse and should be avoided for run-level choices.

The reasoning

Airflow 2 built a `CronDataIntervalTimetable` from a cron string: a run for Monday's interval fired on Tuesday, after the interval ended, with `data_interval_start` on Monday. Airflow 3 builds a `CronTriggerTimetable` instead: the run fires at the cron time and its logical date is that moment. `[scheduler] create_cron_data_intervals` restores the old behaviour.

A `timedelta` builds a `DeltaDataIntervalTimetable`, runs spaced by the delta from the start date with real intervals. `@continuous` runs the DAG again as soon as the previous run finishes, with `max_active_runs=1`. `@once` runs exactly one time at start_date. None means only manual or API triggers.

A custom Timetable subclass implements `next_dagrun_info` and `infer_manual_data_interval`, which is how trading-day or last-business-day schedules are built. It is registered through a plugin so the dag processor can deserialise it. `EventsTimetable` takes an explicit list of datetimes.

Assets, formerly Datasets, are the fourth mechanism: `schedule=[Asset('s3://bucket/orders')]` runs the DAG when a producing task updates that asset. Airflow 3 also allows combining assets with a time schedule through `AssetOrTimeSchedule`.

The reasoning

`depends_on_past=True` on `load` means Tuesday's `load` waits for Monday's `load` to succeed. `wait_for_downstream=True` on `load` means Tuesday's `load` also waits for Monday's `transform`, if `transform` is directly downstream of `load`. Only the immediate children count, not the whole run.

The case is a shared mutable target. `load` writes a staging table, `transform` reads it. Without the setting, Tuesday's `load` can overwrite staging while Monday's `transform` is mid-read. With it, Tuesday's `load` waits until Monday's read finished.

Both settings serialise runs and reduce throughput, and both can stall a DAG forever: a failed Monday blocks every later day until it is fixed or marked success. The first run has no past and is exempt.

The cleaner fix is to remove the shared resource: write per-run partitions keyed by logical date so runs never touch the same table. Then neither setting is needed and `max_active_runs` can be above one.

The reasoning

`on_failure_callback=notify_slack` on a task, or in default_args for all tasks. The function signature is `def notify_slack(context):`, and `context['task_instance']`, `context['exception']` and `context['dag_run']` give the message its content. A list of callables is accepted.

`on_retry_callback` fires on each failed attempt that has retries left; `on_failure_callback` fires once, after the last. Alerting on retry produces noise; alerting on failure is the usual choice. `on_execute_callback` is for audit logging at start.

DAG-level `on_failure_callback` and `on_success_callback` fire when the run finishes in that state and historically ran in the scheduler, so a slow callback slowed scheduling. Keep callbacks fast and fail-safe; an exception inside one is logged and swallowed.

SLAs, a time by which a task should finish measured from the logical date, were unreliable and were removed in Airflow 3.0. Airflow 3.1 introduced deadline alerts, defined on the DAG with a reference time and a callback. Many teams instead alert from outside on the data: has the target table been updated by nine.

The reasoning

`TriggerDagRunOperator(trigger_dag_id='load_b', conf={'date': '{{ ds }}'}, wait_for_completion=True)` creates a manual run of `load_b` with that conf. With `wait_for_completion`, it polls until the run finishes and fails if the run fails. `logical_date` can be set so the child run carries the parent's date.

`ExternalTaskSensor(external_dag_id='extract_a', external_task_id='done')` succeeds when that task in the run with the same logical date reaches success. If the DAGs run at different times, `execution_delta` or `execution_date_fn` shifts the match; `allowed_states` and `failed_states` tune it. It is a sensor, so poke and reschedule modes and timeouts apply.

The sensor's weakness is the date match: a DAG on a different schedule, or a manual run with an arbitrary logical date, never matches and the sensor waits until timeout. The trigger's weakness is coupling: A must know B exists, and B cannot run on its own schedule.

Assets, called Datasets before Airflow 3, invert the dependency. The producer declares `outlets=[Asset('s3://bucket/orders')]`; the consumer sets `schedule=[Asset('s3://bucket/orders')]`. No logical dates to align, and the asset view in the UI draws the lineage.

The reasoning

`[secrets] backend = airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend` with `backend_kwargs` giving the prefixes, such as `connections_prefix = airflow/connections`. A connection `dwh` is then read from the secret `airflow/connections/dwh`, holding a URI or a JSON document.

Environment variables are the second source: `AIRFLOW_CONN_DWH` and `AIRFLOW_VAR_REGION`, in URI or JSON form. They are how containers inject credentials without a backend. The metadata database, where the UI's Admin pages write, is checked last.

The order explains the surprises. A connection edited in the UI has no effect if the backend defines the same id. A variable in the environment is invisible to the UI's list, because the UI shows only the database. Secrets from a backend are never written to the database.

Every lookup is a call to the store, so a DAG file that reads `Variable.get` at module level hits the backend on every parse. Read secrets inside tasks, and cache where the backend supports it. In Airflow 3 the worker asks the API server, which performs the lookup, so workers need no store credentials.

The reasoning

`BashOperator(task_id='say_hi', bash_command='echo hi')` is a task built from the BashOperator class. The class defines what happens; the task_id and arguments define this use of it. The same class appears as many tasks.

Operators subclass BaseOperator, which supplies retries, trigger rules, pools, templating and callbacks. A custom operator overrides `execute(self, context)` and lists its `template_fields`. Most operators wrap a hook for the external system.

Families: action operators that do work, transfer operators that copy data between systems, and sensors that wait. In Airflow 3 the core ones come from the standard provider and the rest from provider packages.

The `@task` decorator is the function form of the PythonOperator; it is still an operator underneath, and the function's return value becomes an XCom.

The reasoning

Each state has a colour in the grid. Green success, red failed, orange up_for_retry, pink skipped, dark red upstream_failed, purple deferred. Reading a failed run means reading which task reached which state first.

`scheduled` means dependencies are met and a slot is being sought. `queued` means the executor has it and a worker has not started it yet; a long time in queued points at workers or pools. `deferred` means the task is waiting on the triggerer with no worker slot held.

`upstream_failed` is the state of a task whose parent failed and whose trigger rule is all_success; it never ran. `skipped` comes from branching or a short circuit. `removed` means the task no longer exists in the DAG file for a run that had it, common after a mapped list shrank.

Clearing a task instance resets it to none so it reruns, and by default clears its downstream too. Marking success or failed sets the state by hand without running it, which is how a stuck run is unblocked.

The reasoning

The scheduler does not parse DAG files in Airflow 3; the dag processor does, writing serialised DAGs to the database. The scheduler reads those. In Airflow 2 the parsing loop lived inside the scheduler unless the standalone dag processor was enabled.

Creating runs is timetable-driven: for a DAG with `@daily` and catchup, it creates one run per missed day from the start date; with catchup off, only the latest. `max_active_runs` caps how many runs of one DAG are open.

Scheduling tasks means moving instances to scheduled when their trigger rule is satisfied, then to queued when a pool slot and the DAG's concurrency allow, ordered by priority weight. The executor then runs them.

High availability came in 2.0: multiple schedulers use row-level locks in PostgreSQL to share the work. Scheduler health is the first thing to check when nothing runs, and the `scheduler_heartbeat` metric is what monitoring watches.

The reasoning

The executor lives inside the scheduler process. It receives tasks the scheduler has queued and places them. It does not decide what runs; the scheduler does. It decides on which machine and in what isolation.

LocalExecutor is the single-machine answer: parallel subprocesses, no queue, simple, limited by one host. CeleryExecutor adds Redis or RabbitMQ and long-lived workers, giving horizontal scale with warm processes and fast task start. Queues route tasks to worker groups.

KubernetesExecutor gives each task its own pod with its own image and resources, isolating dependencies and scaling to zero, at the cost of pod start-up latency for every task. Many deployments use Celery for short tasks and KubernetesPodOperator for heavy ones, or run both executors since 2.10.

SequentialExecutor, which ran one task at a time against SQLite, was removed in 3.0. Managed services wrap these: MWAA and Cloud Composer run Celery or Kubernetes executors; Astronomer offers both. The choice is operational, not about DAG code.

The reasoning

Sensors subclass BaseSensorOperator and implement `poke(context) -> bool`. Common ones: S3KeySensor, GCSObjectExistenceSensor, SqlSensor, HttpSensor, ExternalTaskSensor, DateTimeSensor, FileSensor. `@task.sensor` builds one from a function.

Defaults are poke mode, a 60-second interval and a 7-day timeout. A sensor left on defaults for a file that never arrives occupies a slot for a week. Set `timeout` to the real deadline, and `soft_fail=True` if a missing input should skip rather than fail.

`mode='reschedule'` frees the slot between pokes. `deferrable=True`, where the sensor supports it, moves the wait to the triggerer with no slot at all. Both are the production choice.

Sensors are also a design smell when overused. A DAG scheduled on an asset, or triggered by the producer, needs no sensor. Reach for a sensor when the producer is outside Airflow.

The reasoning

The key design property is that a run for a past date processes that date's data, because the task reads the logical date from context and not the wall clock. Backfill is only safe when tasks are idempotent for their date: rerunning 3 March produces the same 3 March partition.

`catchup` defaults to false in Airflow 3, so a new DAG with a start_date in the past does not flood the system with runs. With catchup true, pausing a DAG for a week produces seven runs on unpause. Catchup is automatic; a backfill is a deliberate act with a chosen range.

In Airflow 2 `airflow dags backfill` ran the DAG in a separate process on the command line, outside the scheduler, with its own problems. Airflow 3 moved backfills into the scheduler: `airflow backfill create --dag-id orders --from-date 2026-01-01 --to-date 2026-01-31`, with `--max-active-runs` and a reprocess behaviour for dates that already have runs.

Backfills compete with production for pools and warehouse capacity, so they are limited with max active runs, run in a separate pool, or pointed at a separate target. A backfill of a non-idempotent DAG duplicates data; that is the first question to ask before running one.

The reasoning

`conn_id='dwh'` with type postgres, host, port 5432, login, password and schema. `S3Hook(aws_conn_id='aws_default')` or `PostgresHook(postgres_conn_id='dwh')` reads it. The DAG never holds a password.

The URI form is `postgresql://user:pass@host:5432/db`; the JSON form, preferred since 2.3, is `{"conn_type": "postgres", "host": ..., "extra": {...}}`. Either can be the value of `AIRFLOW_CONN_DWH`. The CLI is `airflow connections add`.

The `extra` field holds type-specific settings as JSON: a role and warehouse for Snowflake, a region for AWS. Provider hooks define which extra keys they read and render a form for them in the UI.

Passwords in the database are encrypted with the Fernet key from the config; losing the key makes them unreadable. Production deployments put connections in a secrets backend and leave the database empty of credentials. Connection test from the UI calls the hook's `test_connection`.

The reasoning

`POST /api/v2/dags/{dag_id}/dagRuns` with a JSON body containing `conf` and optionally `logical_date` starts a run; that is how an event from outside Airflow becomes a pipeline run. `GET /api/v2/dags/{dag_id}/dagRuns/{run_id}/taskInstances` reads progress.

Airflow 2 served `/api/v1` from the webserver with basic auth or a configured backend. Airflow 3 serves `/api/v2` from the API server; the client first calls `/auth/token` with credentials to receive a JWT and sends it as a bearer token. The auth manager plugin decides who is allowed what.

Airflow 3 also has a separate internal task execution API, used by workers for XComs, variables and connections. It is not for external callers; the public API is v2.

Common uses: a CI pipeline pausing a DAG during deploy, a data platform portal showing pipeline status, and an external scheduler triggering a run after a file lands. The Python client `apache-airflow-client` is generated from the OpenAPI spec.

The reasoning

`KubernetesPodOperator(task_id='train', image='myrepo/trainer:1.4', cmds=['python', 'train.py'], arguments=['--date', '{{ ds }}'], get_logs=True)`. The pod runs to completion, logs stream into the task log, and the task state follows the pod's exit code.

It isolates dependencies completely: the trainer image needs no Airflow installed. It is the standard way to run Spark submit, dbt, or any container from Airflow. `do_xcom_push=True` collects a result the container writes to `/airflow/xcom/return.json` via a sidecar.

`on_finish_action` controls pod cleanup, `container_resources` sets requests and limits, and `kubernetes_conn_id` points at the cluster, which can be a different cluster from the one Airflow runs on. Pod templates allow full specs.

You can use the operator with LocalExecutor or CeleryExecutor: the worker only launches and watches the pod. With KubernetesExecutor plus the operator, there are two pods per task, the Airflow task pod and the work pod, which is normal and sometimes surprising.

The reasoning

`BashOperator(bash_command='load.sh {{ ds }}')` runs with yesterday's or today's date depending on the timetable, rendered at run time. `SQLExecuteQueryOperator(sql='delete from t where d = \'{{ ds }}\'')` renders the SQL. Files ending in `.sql` or `.sh` named in a templated field are read and rendered too, controlled by `template_ext`.

The context holds `logical_date`, `data_interval_start`, `data_interval_end`, `ds`, `ds_nodash`, `ts`, `dag_run`, `ti`, `task`, `params`, `run_id` and more. `{{ ti.xcom_pull(task_ids='extract') }}` reads an XCom in a template. `{{ var.value.region }}` and `{{ conn.dwh.host }}` read a Variable and a connection.

Macros: `{{ macros.ds_add(ds, -1) }}` shifts a date; `{{ macros.ds_format(ds, '%Y-%m-%d', '%Y%m%d') }}` reformats; `macros.datetime` and `macros.timedelta` expose the modules. `user_defined_macros` and `user_defined_filters` on the DAG add your own.

A custom operator declares `template_fields = ('sql', 'path')` so its arguments render. In Airflow 3, `execution_date`, `next_ds`, `prev_ds` and the other deprecated names are gone. Code must use `logical_date` and the data interval names, and `logical_date` can be None for a manually triggered run.