Workflow Orchestration

A DAG (Directed Acyclic Graph) in Apache Airflow represents a workflow — a collection of tasks with defined execution order and no cycles. Key parameters include dag_id (unique identifier), schedule (cron expression or timedelta), start_date (first logical date), catchup (whether to backfill missed runs, default True), max_active_runs (concurrency limit), tags (UI labels), and default_args (task-level defaults like retries and owner). DAGs are discovered by the scheduler from Python files in the DAGS_FOLDER. In Airflow 2.x, the @dag decorator is preferred over the DAG context manager for cleaner code.

@dag(schedule='0 6 * * *', start_date=datetime(2025, 1, 1), catchup=False)
def fraud_pipeline():
    pass

The TaskFlow API (introduced in Airflow 2.0) uses Python decorators to define tasks and DAGs, eliminating boilerplate operator instantiation and explicit XCom push/pull calls. Decorate a Python function with @task to make it an Airflow task; return values automatically become XComs consumed by downstream tasks as function arguments. Use @dag to define the DAG function. TaskFlow makes dependencies implicit and readable — calling task_b(task_a()) defines a dependency — versus explicit set_downstream() calls. It integrates with type hints and makes unit testing easier since tasks are plain Python functions. TaskFlow is the recommended approach for new Airflow 2.x DAGs.

@dag(schedule='@daily', start_date=datetime(2025,1,1))
def my_pipeline():
    @task
    def extract(): return {'rows': 1000}
    @task
    def load(data): print(data['rows'])
    load(extract())

XCom (cross-communication) is Airflow's mechanism for passing small data between tasks. Tasks push values with xcom_push(key, value) and pull with xcom_pull(task_ids, key). In TaskFlow API, return values and function arguments handle this transparently. XComs are stored in the Airflow metadata database, which imposes size limits: SQLite allows 2 GB but practical limits are much smaller; PostgreSQL BYTEA allows ~1 GB. The recommended maximum is a few MB; for larger data, store results in external storage (S3, GCS) and pass only the path as XCom. The XCOM_BACKEND setting allows customizing XCom storage to external systems in Airflow 2.x.

# TaskFlow auto-XCom
@task
def extract() -> str:
    return 's3://bucket/data.parquet'  # pass path, not data

@task
def transform(path: str):
    df = pd.read_parquet(path)

Airflow executors determine how tasks are dispatched and run. SequentialExecutor runs one task at a time in the scheduler process (development only). LocalExecutor runs tasks in subprocesses on the scheduler machine. CeleryExecutor distributes tasks to a pool of persistent Celery workers connected via a message broker (Redis or RabbitMQ), with workers running continuously and requiring pre-installed dependencies. KubernetesExecutor spins up a fresh Kubernetes pod per task, providing perfect isolation and dynamic resource allocation, but with higher pod startup latency (10–30 seconds). CeleryExecutor offers lower task latency; KubernetesExecutor offers better isolation and Kubernetes-native scaling.

# airflow.cfg
[core]
executor = KubernetesExecutor

[kubernetes]
namespace = airflow
worker_container_repository = myrepo/airflow
worker_container_tag = latest

Trigger rules define when a task is eligible to run based on the states of its upstream tasks, overriding the default all_success behavior. all_success (default): run when all upstreams succeed. all_failed: run when all upstreams fail (useful for cleanup tasks). all_done: run regardless of upstream states as long as all are finished. one_success: run if at least one upstream succeeds. one_failed: run if at least one upstream fails. none_failed: run if no upstreams failed (skipped is acceptable). none_skipped: run if no upstreams are skipped. Trigger rules are set per task: task = PythonOperator(..., trigger_rule='all_done').

cleanup = PythonOperator(
    task_id='cleanup',
    python_callable=cleanup_fn,
    trigger_rule='all_done'  # runs even if upstreams failed
)

BranchPythonOperator enables conditional branching in a DAG. The python_callable must return a task_id string (or list of task_ids for multiple branches) representing which downstream tasks should execute. Tasks on non-selected branches are marked as skipped. Downstream tasks of skipped branches are also skipped by default unless their trigger_rule overrides this (e.g., none_failed_min_one_success). In TaskFlow API, use the @task.branch decorator equivalently. A common pattern gates downstream processing on data quality checks: if data is valid, run transform; otherwise run send_alert. The branch function receives the Airflow context dictionary as its argument.

@task.branch
def check_quality(row_count: int) -> str:
    return 'transform' if row_count > 0 else 'send_alert'

branch = check_quality(count)
transform_task = transform()
alert_task = send_alert()
branch >> [transform_task, alert_task]

Airflow Connections store external system credentials (database URIs, API keys, cloud credentials) identified by a conn_id string. They are retrieved by Hooks using BaseHook.get_connection(conn_id). Store connections in the Airflow UI (Admin > Connections), environment variables (AIRFLOW_CONN_<CONN_ID_UPPER>=uri), or a Secrets Backend such as AWS Secrets Manager, HashiCorp Vault, or GCP Secret Manager. The Secrets Backend is recommended for production to avoid storing credentials in the metadata database. The connection URI format is: conn_type://login:password@host:port/schema. Airflow 2.x supports JSON-encoded connections for complex credentials with multiple fields.

# Environment variable connection (preferred for secrets)
export AIRFLOW_CONN_POSTGRES_PROD='postgresql://user:pass@db-host:5432/prod'

A Sensor is a special operator that polls for a condition (file arrival, S3 key existence, external DAG completion, SQL row presence) before allowing downstream tasks to proceed. Traditional sensors occupy a worker slot while polling, wasting resources. Deferrable operators (Airflow 2.2+) address this: they suspend execution, release the worker slot, and register a trigger with the Triggerer component. When the condition is met, the trigger fires and the task resumes on a worker. Use deferrable variants (e.g., TimeSensorAsync, S3KeySensorAsync) for long-wait conditions to dramatically reduce worker slot consumption and improve cluster efficiency.

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensorAsync
wait_for_file = S3KeySensorAsync(
    task_id='wait_s3',
    bucket_name='data-lake',
    bucket_key='input/{{ds}}/data.parquet'
)

Airflow's scheduler uses the schedule parameter and start_date to determine run timing. Crucially, a DAG run with logical_date D runs after interval D is complete — for a daily DAG with start_date 2025-01-01, the first run has logical_date 2025-01-01 but is triggered at the end of that day (2025-01-02 00:00). This data interval semantics means a daily DAG always processes yesterday's data. The scheduler scans DAG files continuously (dag_file_processor_timeout), creates DagRuns for each missed interval if catchup=True, and queues TaskInstances for the executor. max_active_runs limits concurrent DagRuns per DAG to prevent overloading.

@dag(
    schedule='@daily',
    start_date=datetime(2025, 1, 1),
    catchup=False,
    max_active_runs=3
)

Data-aware scheduling (Datasets API, Airflow 2.4+) allows DAGs to trigger based on the completion of tasks in other DAGs that produce named datasets, rather than time-based schedules. Define a Dataset with a URI (e.g., Dataset('s3://bucket/users.parquet')). Mark producing tasks with outlets=[dataset] and consuming DAGs with schedule=[dataset]. The scheduler triggers the consumer DAG each time all listed datasets are updated by their producers. This replaces fragile ExternalTaskSensor patterns with an explicit producer-consumer dependency graph visible in the Airflow UI as a Dataset dependency view, enabling event-driven pipeline orchestration.

from airflow.datasets import Dataset
users_ds = Dataset('s3://lake/users.parquet')

# Producer DAG
@task(outlets=[users_ds])
def build_users(): ...

# Consumer DAG
@dag(schedule=[users_ds])
def downstream_pipeline(): ...

Dynamic task mapping generates a variable number of parallel task instances at runtime based on input data, replacing static task fan-out patterns. Use task.expand(param=list_of_values) or task.expand_kwargs(list_of_dicts) to create mapped task instances. Each list element runs as an independent parallel task instance. The downstream task receives a list of all mapped results. The number of instances is determined at runtime, enabling data-driven parallelism without hardcoded task counts. Combine with partial() to fix some parameters while mapping others. Dynamic mapping works with both TaskFlow @task functions and traditional operators via partial().expand().

@task
def process_file(filename: str) -> int:
    return len(open(filename).readlines())

files = get_files()  # returns ['a.csv', 'b.csv', 'c.csv']
process_file.expand(filename=files)

Airflow Variables are key-value pairs stored in the metadata database for passing configuration values to DAGs (e.g., bucket names, feature flags, thresholds). Access them with Variable.get('key', default_var='default'). Sensitive variables can be encrypted at rest using Airflow's Fernet key. Variables are fetched at DAG parse time if called outside task functions, causing scheduler performance issues — always access Variables inside task callables or use jinja templating ({{ var.value.key }}) to defer fetching to runtime. Connections, in contrast, store structured credential objects (host, port, login, password, schema) for database and API authentication. Both support Secrets Backend integration.

@task
def fetch():
    bucket = Variable.get('s3_bucket', default_var='default-bucket')
    # Access inside task to avoid scheduler-time fetching

KubernetesPodOperator launches a Kubernetes pod to execute a task, giving each task its own container image, resource limits, environment variables, volume mounts, and service account. The pod runs to completion and is deleted afterward (unless on_finish_action='keep_pod'). Use KubernetesPodOperator when tasks require custom Docker images with specific dependencies, GPU resources, isolated secrets, or large memory allocations incompatible with shared Airflow workers. It is more flexible than KubernetesExecutor (which uses a fixed image) and enables polyglot pipelines where different tasks run different language runtimes. Requires a Kubernetes cluster and appropriate RBAC permissions.

from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
train_task = KubernetesPodOperator(
    task_id='train_model',
    image='myrepo/trainer:v2',
    arguments=['python', 'train.py'],
    resources={'request_memory': '8Gi', 'limit_gpu': '1'}
)

Pools are named resource slots that limit the number of concurrently running tasks sharing a scarce resource (e.g., database connections, API rate limits, GPU slots). Create pools in the UI (Admin > Pools) with a defined number of slots. Assign tasks to pools with pool='pool_name'. Tasks wait in a queue until a slot is available. priority_weight controls which queued tasks get slots first. Pool slots are distinct from Airflow's overall parallelism setting — a task must satisfy both parallelism limits and pool availability. Use pools to prevent overloading external systems: limit Snowflake queries to 10 concurrent tasks, or restrict GPU training tasks to 4 slots matching physical GPU count.

gpu_task = PythonOperator(
    task_id='train',
    python_callable=train_fn,
    pool='gpu_pool',        # limit to pool slots
    priority_weight=10      # higher = preferred
)

TaskGroups (introduced in Airflow 2.0) visually and logically group related tasks in the Airflow UI graph view, collapsing into a single node and expanding on click. They replace SubDAGs, which had significant performance and concurrency issues. Define groups with the TaskGroup context manager or @task_group decorator. Groups can be nested. Dependencies between groups are set using the group object itself — group_a >> group_b — and Airflow expands these to all tasks in the groups. TaskGroups are a UI-only organizational construct with no execution semantics change, avoiding the deadlock and concurrency issues of SubDAGs which ran as separate DagRuns.

from airflow.utils.task_group import TaskGroup
with TaskGroup('feature_engineering') as fe_group:
    normalize = PythonOperator(task_id='normalize', ...)
    encode = PythonOperator(task_id='encode', ...)
extract >> fe_group >> train

catchup=True (default) instructs the scheduler to create DAG runs for all missed schedule intervals between start_date and the current date when a DAG is unpaused or first deployed. This can flood the system with thousands of queued runs. Set catchup=False to run only from the most recent interval, skipping historical gaps. For explicit backfilling, use the airflow dags backfill -s START_DATE -e END_DATE command regardless of catchup setting. Backfills respect max_active_runs but run in parallel within that limit. Use depends_on_past=True on individual tasks to enforce sequential backfill execution when downstream tasks depend on previous day's results.

@dag(schedule='@daily', start_date=datetime(2025,1,1), catchup=False)
def pipeline(): ...

# Manual backfill via CLI:
# airflow dags backfill -s 2025-01-01 -e 2025-03-31 pipeline

SLAs in Airflow define a maximum allowed duration for a task or DAG run to complete. Set sla=timedelta(hours=2) on a task to trigger an SLA miss callback if the task has not finished within 2 hours of the DAG's scheduled execution time (not the task start time). Define the sla_miss_callback function on the DAG object to receive notifications (e.g., send a Slack message or PagerDuty alert) when SLAs are breached. SLA misses are recorded in the sla_miss table in the metadata database and visible in the UI. Note: SLA checks run on scheduler heartbeats, so there can be a delay between the breach and the notification. SLAs do not kill tasks — use execution_timeout for hard cutoffs.

def sla_miss_handler(dag, task_list, blocking_task_list, slas, blocking_tis):
    send_slack_alert(f'SLA missed: {[t.task_id for t in task_list]}')

@dag(sla_miss_callback=sla_miss_handler, ...)
def pipeline():
    task = PythonOperator(..., sla=timedelta(hours=1))

Hooks are thin abstractions over external service clients, handling connection retrieval and authentication transparently using a conn_id. They encapsulate the raw API calls for databases (PostgresHook, MySqlHook), cloud services (S3Hook, GCSHook, BigQueryHook), HTTP APIs (HttpHook), and more. Operators typically create a Hook internally using a conn_id parameter and delegate actual work to it. This separation enables reuse of connection logic across operators and custom tasks. Create hooks directly in @task functions for lightweight access without needing a full operator. Hooks also manage connection pooling and cleanup, ensuring resources are released after task completion.

@task
def load_to_postgres(data: list):
    hook = PostgresHook(postgres_conn_id='prod_db')
    hook.insert_rows(table='events', rows=data)

Configure retries using default_args or per-task parameters. retries sets the number of retry attempts; retry_delay=timedelta(minutes=5) sets the initial wait between attempts. Enable exponential backoff with retry_exponential_backoff=True, which doubles the delay on each retry (min_backoff_secs configures the floor). Set retry_on_exception(exception_type) at the task level in newer providers to limit retries to specific exception types. Use on_retry_callback to execute alerting logic on each retry. Hard-limit total task runtime with execution_timeout=timedelta(hours=2), which raises an AirflowTaskTimeout exception regardless of retries. These settings can be defined in default_args to apply globally across all DAG tasks.

default_args = {
    'retries': 3,
    'retry_delay': timedelta(minutes=2),
    'retry_exponential_backoff': True,
    'execution_timeout': timedelta(hours=1)
}

ExternalTaskSensor waits for a task or entire DAG in another DAG to reach a specified state (default: success) before allowing the dependent task to proceed. Key parameters: external_dag_id (target DAG), external_task_id (target task, or None to wait for the whole DAG run), allowed_states (['success'] by default), execution_date_fn (maps the current DAG's logical date to the target DAG's date — critical when schedules differ), timeout, and poke_interval. For Airflow 2.4+, prefer Datasets-based scheduling over ExternalTaskSensor when the dependency relationship is a data production event, as Datasets provide explicit, observable producer-consumer coupling.

ExternalTaskSensor(
    task_id='wait_for_upstream',
    external_dag_id='data_ingestion',
    external_task_id='validate',
    execution_date_fn=lambda dt: dt,
    timeout=3600
)

For payloads exceeding safe XCom size (a few KB to MB), use the object-store-pass-through pattern: serialize the data to a remote store (S3, GCS, or a database) and pass only the URI or key as an XCom string. In Airflow 2.x, configure a custom XCom backend by implementing BaseXCom and setting xcom_backend in airflow.cfg. AWS users can set the XCom backend to store values in S3 automatically, making large XComs transparent. Alternatively, use shared NFS or database tables and pass table names. Avoid passing DataFrames, model weights, or large lists directly through XCom — it bloats the metadata database and creates performance issues for the scheduler.

@task
def process() -> str:
    df = compute_large_df()
    path = 's3://bucket/tmp/result.parquet'
    df.to_parquet(path)
    return path  # only the path is XCom'd

@task
def consume(path: str):
    df = pd.read_parquet(path)

Airflow 2.x provides a stable REST API (enabled by default) at /api/v1/ for programmatic DAG management. Authenticate with Basic Auth, Kerberos, or OAuth2 (configurable via AUTH_BACKEND). Trigger a DAG run with POST /api/v1/dags/{dag_id}/dagRuns, passing a JSON body with logical_date, conf (runtime configuration dictionary), and run_id. The conf dict is accessible inside tasks via context['dag_run'].conf. Other key endpoints: GET /dags (list DAGs), PATCH /dags/{dag_id} (pause/unpause), GET /dags/{dag_id}/dagRuns (list runs), GET /taskInstances (query task states). Use the REST API in CI/CD pipelines to trigger training DAGs on code merges.

curl -X POST http://airflow:8080/api/v1/dags/train_model/dagRuns \
  -H 'Content-Type: application/json' \
  -u admin:admin \
  -d '{"conf": {"model_version": "v3", "dataset": "2025-Q4"}}'

CeleryExecutor uses a message broker (Redis or RabbitMQ) as a task queue and persistent Celery workers that pull tasks from the queue. The Airflow scheduler submits TaskInstance commands to the broker. Workers pick up tasks matching their configured queues. Multiple workers with different queue subscriptions enable task routing — e.g., a gpu_queue for GPU tasks, a cpu_queue for data processing. Workers must have Airflow and all DAG dependencies installed. Worker count scales independently of the scheduler. The broker and Celery result backend (usually Redis or PostgreSQL) must be highly available. Monitor workers with Celery Flower or the Airflow UI Worker tab.

# Route specific tasks to GPU workers
gpu_task = PythonOperator(
    task_id='train',
    python_callable=train,
    queue='gpu'  # workers subscribed to 'gpu' queue handle this
)

Setup and teardown tasks are a declarative pattern for resource lifecycle management. A setup task (marked with .as_setup()) provisions infrastructure (spins up an EMR cluster, creates a database schema). A teardown task (marked with .as_teardown()) deprovisions it. Teardown tasks run regardless of pipeline success or failure — similar to trigger_rule=all_done but with explicit semantic meaning. Link setup and teardown to work tasks using the >> notation. If the setup fails, the teardown still runs to prevent resource leaks. In the UI, setup/teardown tasks render distinctly, making infrastructure lifecycle explicit and preventing expensive resource orphaning in failure scenarios.

@task
def create_cluster(): return cluster_id

@task
def delete_cluster(cluster_id): terminate(cluster_id)

setup = create_cluster().as_setup()
teardown = delete_cluster(cluster_id).as_teardown()
setup >> process_data >> teardown

DAG versioning requires care because Airflow serializes DAGs to the database and running DagRuns reference the current DAG structure. Safe versioning strategies: version task_id strings rather than changing existing ones (add _v2 suffix) to avoid orphaned TaskInstances. Use dag_id versioning (append a version suffix) for breaking changes that require a clean DAG history. Maintain backward-compatible task_id names for in-flight runs. Deploy new DAG files atomically to the DAGS_FOLDER using atomic file moves or git-based synchronization (gitSync in Kubernetes). Serialize DAGs using Airflow's DAG serialization feature to reduce scheduler parsing overhead. Test DAG changes in a staging Airflow environment before production deployment.

DAG serialization (enabled by default in Airflow 2.0+) stores a JSON representation of each DAG in the metadata database after parsing. The scheduler only needs to re-parse DAG Python files when they change, then updates the serialized form. The webserver reads DAG structure from the database rather than parsing Python files on each UI request, reducing parse overhead and enabling the webserver to run without access to the DAG files. This significantly reduces scheduler CPU usage on large installations with hundreds of DAGs. Serialized DAGs are stored in the serialized_dag table. Configure min_serialized_dag_update_interval to control how frequently changed DAGs are re-serialized.

# airflow.cfg
[core]
min_serialized_dag_update_interval = 30
min_serialized_dag_fetch_interval = 10

Idempotent tasks produce the same result regardless of how many times they are run with the same inputs. Implement idempotency by: using upsert patterns instead of INSERT for database writes (INSERT ... ON CONFLICT DO UPDATE); deleting and recreating output files or table partitions before writing; using atomic write patterns (write to temp path, then rename); adding WHERE clauses that check for existing records before insertion; using Airflow's logical_date to scope operations to a specific partition. Idempotency is critical for safe retries and backfills — Airflow can re-run any task at any time, so non-idempotent tasks cause duplicate data or incorrect aggregations in production.

@task
def write_partition(**context):
    dt = context['ds']  # logical date as 'YYYY-MM-DD'
    # Delete before insert = idempotent partition overwrite
    spark.sql(f"ALTER TABLE events DROP IF EXISTS PARTITION (dt='{dt}')")
    df.write.partitionBy('dt').mode('append').parquet(path)

ShortCircuitOperator evaluates a Python callable that returns True or False. If True, downstream tasks proceed normally. If False, all downstream tasks are marked as skipped, short-circuiting the remainder of the pipeline without failure. It is useful for conditional execution based on runtime data: skip expensive model retraining if metrics haven't degraded, skip a notification task if no anomalies were found, or skip a load task if the data quality check failed. The ignore_downstream_trigger_rules parameter (default True) controls whether downstream trigger_rule settings are respected or overridden during skipping. In TaskFlow, use @task.short_circuit equivalently.

@task.short_circuit
def should_retrain(drift_score: float) -> bool:
    return drift_score > 0.15  # only retrain if drift detected

should_retrain(score) >> retrain_model >> register_model

Test DAG structure with Python import: python my_dag.py should raise no errors. Check DAG integrity with airflow dags list and airflow dags test <dag_id> <execution_date> which dry-runs the entire DAG without the scheduler. Test individual tasks with airflow tasks test <dag_id> <task_id> <execution_date> which runs the task synchronously in the local process. Unit test @task functions as plain Python functions by calling them directly. Use pytest-airflow or the DAG.test() method (Airflow 2.5+) for integration-style testing with in-memory databases. Mock external connections using unittest.mock.patch on Hook classes. Keep task functions pure and testable by separating business logic from Airflow-specific code.

# CLI task test
airflow tasks test fraud_pipeline validate_data 2025-01-15

# DAG.test() in pytest
def test_dag():
    dag = fraud_pipeline()
    dag.test(run_conf={'env': 'test'})

LocalExecutor runs tasks as subprocesses on the same machine as the Airflow scheduler, using Python's multiprocessing module. It supports parallelism (multiple tasks run concurrently as separate OS processes), unlike SequentialExecutor which runs one task at a time. LocalExecutor is appropriate for small to medium workloads on a single powerful machine: development environments, simple production pipelines with modest task counts, or situations where managing a Celery or Kubernetes cluster is not justified. Resource consumption is limited to the scheduler machine's CPU and memory. Maximum parallelism is configured with the parallelism setting. For large-scale or heterogeneous task requirements, migrate to CeleryExecutor or KubernetesExecutor.

# airflow.cfg
[core]
executor = LocalExecutor
parallelism = 32

Airflow stores all datetimes as UTC internally. Set default_timezone in airflow.cfg to display dates in a preferred time zone in the UI. DAG start_date and schedule arguments should always be timezone-aware or expressed in UTC. Use pendulum (the built-in library) for timezone-aware datetimes: pendulum.datetime(2025, 1, 1, tz='US/Eastern'). A common pitfall is setting a naive (timezone-unaware) datetime for start_date, which Airflow treats as UTC, causing unexpected first-run timing. Cron expressions in schedule are always interpreted in UTC. When scheduling relative to business hours, use TimetableCRUD plugins or CustomTimetable subclasses to express timezone-aware schedules that respect DST.

import pendulum
@dag(
    schedule='30 9 * * 1-5',  # 9:30 UTC on weekdays
    start_date=pendulum.datetime(2025, 1, 1, tz='UTC'),
    catchup=False
)

depends_on_past=True means a task instance will not run unless the same task in the previous DAG run has succeeded. This creates sequential ordering across runs — run N waits for run N-1's same task to finish successfully. Useful for rolling computations where today's output depends on yesterday's (e.g., cumulative aggregations). The implication is that a failure in one run blocks all future runs of that task, requiring manual intervention to clear the failed instance before the pipeline resumes. The first DAG run's task ignores depends_on_past. Avoid this setting unless the data dependency is genuine, as it serializes otherwise parallelizable backfills and increases fragility.

cumulative_agg = PythonOperator(
    task_id='cumulative_agg',
    python_callable=compute_cumulative,
    depends_on_past=True  # waits for yesterday's same task
)

Airflow Params (enhanced in 2.2+) allow DAGs to accept typed, validated runtime configuration with UI form rendering. Define Param objects in the params dictionary of the @dag decorator or DAG constructor. Types include STRING, INTEGER, NUMBER, BOOLEAN, ARRAY, OBJECT. Params have default values, descriptions, and JSON Schema validation. In the Trigger DAG UI, a dynamic form is rendered from the schema. Access param values in tasks via context['params']['key'] or using Jinja templates {{ params.key }}. Params replace unstructured conf dictionaries with type-safe, self-documenting configuration that validates inputs before the DAG run starts.

from airflow.models.param import Param
@dag(
    params={
        'model_version': Param('v1', type='string'),
        'epochs': Param(10, type='integer', minimum=1)
    }
)
def train_dag(**context):
    ver = context['params']['model_version']

Airflow emits StatsD metrics (configurable via statsd_on=True) covering scheduler heartbeats, task durations, DAG processing times, executor queue depths, and task success/failure counts. Forward these to Prometheus via the statsd_exporter sidecar and visualize in Grafana. Key metrics to alert on: scheduler_heartbeat (gap > 30s indicates scheduler issues), task_duration (exceeding SLA), ti_successes / ti_failures ratio (failure rate), dag_processing_total_parse_time (slow DAG discovery). Configure remote logging to S3/GCS so task logs persist beyond pod lifecycle in Kubernetes deployments. Set log_fetch_timeout_sec and log_fetch_delay_sec to tune log retrieval in the UI.

# airflow.cfg
[metrics]
statsd_on = True
statsd_host = localhost
statsd_port = 8125
statsd_prefix = airflow

For CeleryExecutor workers, use graceful shutdown: send SIGTERM to the worker process, which stops accepting new tasks but completes in-flight ones (celery worker --loglevel=info). Wait for active task count to reach zero (visible in Celery Flower or Airflow Worker tab) before forcibly stopping with SIGKILL. For Kubernetes-based deployments (KubernetesExecutor or KEDA-scaled Celery workers), rolling updates use pod disruption budgets (PDB) to ensure a minimum number of workers remain available during rollout. Never SIGKILL workers with running tasks — tasks will be marked as zombie tasks by the scheduler and eventually failed. Implement proper TERM signal handling in custom task code.

# Graceful Celery worker restart
celery -A airflow.executors.celery_executor worker \
  --concurrency=8 \
  --queues=default,gpu

The Triggerer is an Airflow 2.2+ component that runs an asyncio event loop to manage asynchronous triggers for deferrable operators. When a deferrable task suspends, it registers a trigger object (e.g., TimeDeltaTrigger, FileTrigger) with the Triggerer. The Triggerer runs all registered triggers concurrently in a single asyncio event loop, polling conditions efficiently without holding worker slots. A single Triggerer can handle thousands of concurrent deferred tasks. When a trigger fires (condition met), the Triggerer signals the scheduler, which re-queues the task on a worker to resume execution. Deploy Triggerer alongside the Scheduler and Webserver as a separate Airflow process (airflow triggerer).

# Start Triggerer
airflow triggerer

# Deferrable sensor using Triggerer
from airflow.providers.amazon.aws.triggers.s3 import S3KeyTrigger

Subclass BaseOperator and implement the execute(context) method, which contains the task logic and receives the Airflow context dictionary. Define __init__ with task-specific parameters, calling super().__init__(**kwargs) to pass common operator settings. Declare template_fields as a tuple of attribute names that should support Jinja templating. Override pre_execute() and post_execute() for setup/teardown hooks. Return a value from execute() to automatically push it to XCom. Package the custom operator in a Python module importable from the DAGS_FOLDER or installed as a package. For providers, follow the Airflow provider package structure with provider.yaml and publish to PyPI.

from airflow.models import BaseOperator
class MLTrainOperator(BaseOperator):
    template_fields = ('model_config_path',)
    def __init__(self, model_config_path, **kwargs):
        super().__init__(**kwargs)
        self.model_config_path = model_config_path
    def execute(self, context):
        return train(self.model_config_path)

Airflow excels as a general-purpose workflow orchestrator with robust scheduling, extensive provider integrations, and mature operational tooling. It is well-suited for ML pipelines that mix data engineering, model training, and deployment steps with time-based schedules, external dependencies, and complex branching. Kubeflow Pipelines (KFP) is purpose-built for ML: it uses Kubernetes-native containers per step, has native MLflow/metadata integration, supports ML-specific concepts like pipeline versioning and cached step outputs, and integrates with Katib for hyperparameter tuning. Airflow is preferred for heterogeneous pipelines requiring diverse integrations; KFP for cloud-native ML-centric workflows on Kubernetes where container isolation per step is a first-class requirement.

Zombie tasks are TaskInstances stuck in a RUNNING state after their worker process has died (e.g., worker OOM kill, pod eviction, network partition). Airflow's scheduler detects zombies by checking for heartbeats from running tasks — if no heartbeat arrives within zomby_detection_interval seconds (default 300), the scheduler marks the task as failed and triggers retry logic. The scheduler_zombie_task_threshold setting controls the timeout. In KubernetesExecutor, pod failures are detected via the Kubernetes API rather than heartbeats. Manual recovery: use the Airflow UI to Clear a failed TaskInstance (resets it to None state so the scheduler re-queues it) or mark it as success if external execution confirmed completion.

# airflow.cfg tuning
[scheduler]
scheduler_zombie_task_threshold = 300
zombie_detection_interval = 10.0

Airflow providers are separate installable packages (apache-airflow-providers-*) that extend Airflow with operators, hooks, sensors, and connection types for specific services (AWS, GCP, Azure, Databricks, Snowflake, etc.). Install them with pip: pip install apache-airflow-providers-amazon. Providers declare their connection types, operators, and hooks in a provider.yaml file and register them via Python entry points. To add a new cloud service, create a provider package with a Hook subclass handling authentication, Operator subclasses implementing execute(), optional Sensor subclasses, and connection UI widgets. Contributing back to the Apache Airflow monorepo is encouraged. Provider packages version independently from Airflow core.

pip install apache-airflow-providers-amazon
pip install apache-airflow-providers-google
pip install apache-airflow-providers-databricks

Fan-out creates multiple parallel downstream tasks from one task; fan-in waits for multiple upstream tasks to complete before proceeding. In Airflow, set multiple downstream dependencies with task_a >> [task_b, task_c, task_d] for fan-out. For fan-in, set [task_b, task_c, task_d] >> task_e, which means task_e waits for all three with trigger_rule all_success by default. With TaskFlow, calling a task multiple times with different arguments creates parallel invocations (pre-dynamic mapping). For dynamic fan-out with runtime-determined parallelism, use task.expand(). Combine fan-out with pools to limit concurrency of the parallel branches and prevent resource exhaustion.

extract >> [transform_a, transform_b, transform_c]  # fan-out
[transform_a, transform_b, transform_c] >> merge   # fan-in

Use environment-specific configuration via environment variables and Airflow Variables/Connections scoped per environment. Store DAG code in a git repository with a gitSync sidecar (Kubernetes) or CI/CD push to the DAGS_FOLDER. Use Helm charts (official airflow Helm chart) with values files per environment for Kubernetes deployments. Override airflow.cfg settings via AIRFLOW__ prefix environment variables (e.g., AIRFLOW__CORE__EXECUTOR=KubernetesExecutor) without modifying config files. Use separate metadata databases and artifact stores per environment. Implement DAG-level environment detection via Airflow Variables or env vars and route connections accordingly. Infrastructure-as-code (Terraform) manages the Airflow cluster resources per environment.

# Environment variable config override
AIRFLOW__CORE__EXECUTOR=KubernetesExecutor
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql://user:pass@prod-db/airflow
AIRFLOW__LOGGING__REMOTE_BASE_LOG_FOLDER=s3://prod-logs/airflow

max_active_tasks_per_dag (formerly concurrency in older Airflow versions) limits the maximum number of simultaneously running TaskInstances across all active DagRuns of a single DAG. This prevents a single DAG from monopolizing executor resources when it has many parallel tasks or multiple active DagRuns concurrently. Set at the DAG level: @dag(max_active_tasks=16). The global parallelism setting in airflow.cfg caps total concurrent tasks across all DAGs. The relationship is: each DAG is limited by max_active_tasks, all DAGs together are limited by parallelism, and individual tasks within pools are further limited by pool slots. Tune these three settings together for optimal scheduler resource allocation.

@dag(
    schedule='*/15 * * * *',
    max_active_runs=3,
    max_active_tasks=8,  # at most 8 tasks running across all active runs
    catchup=False
)

Configure failure callbacks at three levels. Per-task: on_failure_callback=notify_fn receives the context dict on failure. Per-DAG: on_failure_callback on the DAG object fires when the DAG run fails. Via default_args: apply on_failure_callback to all tasks in a DAG. Callbacks can send Slack messages (SlackWebhookHook), PagerDuty alerts, email (EmailOperator in a separate alert DAG), or custom HTTP notifications. For email, configure SMTP in airflow.cfg (smtp_host, smtp_user, smtp_password) and set email=['team@company.com'] in default_args. Use on_retry_callback for retry notifications. Webhook-based alerts via HttpHook give the most flexibility for integrating with modern incident management platforms.

def slack_alert(context):
    ti = context['task_instance']
    SlackWebhookHook(slack_webhook_conn_id='slack').send(
        text=f'Task {ti.task_id} failed in DAG {ti.dag_id}'
    )
default_args = {'on_failure_callback': slack_alert}

The Airflow webserver provides the browser-based UI and REST API endpoint. It reads DAG structure from the metadata database (serialized DAGs) and task logs from the configured log backend. In production, secure the webserver with FAB (Flask AppBuilder) authentication supporting LDAP, OAuth2 (Google, GitHub, Azure AD), Kerberos, or custom auth backends. Configure AUTH_TYPE in webserver_config.py. Role-based access control (RBAC) restricts users to specific DAGs and actions (Viewer, User, Op, Admin). Deploy the webserver behind a reverse proxy (Nginx, Traefik) for TLS termination, rate limiting, and WAF protection. Do not expose the Airflow webserver directly to the internet.

# webserver_config.py
from flask_appbuilder.security.manager import AUTH_OAUTH
AUTH_TYPE = AUTH_OAUTH
OAUTH_PROVIDERS = [{'name': 'google', 'token_key': 'access_token', ...}]

Airflow orchestrates dbt via the BashOperator (calling dbt run, dbt test, dbt snapshot), the DbtRunOperator from the astronomer-cosmos or dbt-airflow provider packages, or custom Python operators wrapping the dbt Python API. The Astronomer Cosmos library is the most mature integration — it parses dbt project metadata and auto-generates an Airflow DAG with individual task nodes per dbt model, enabling per-model retry, alerting, and dependency visualization in the Airflow UI. This approach replaces a single opaque dbt run command with granular task-level observability. Pass dbt targets, profiles, and vars as Airflow Variables or environment variables to switch environments.

from cosmos import DbtTaskGroup, ProjectConfig, ProfileConfig
DbtTaskGroup(
    project_config=ProjectConfig('/dbt/my_project'),
    profile_config=ProfileConfig(profile_name='prod', target_name='prod', ...)
)

The logical_date (renamed from execution_date in Airflow 2.2) represents the start of the data interval that the DAG run processes, not the actual wall-clock time when the run was triggered. For a daily DAG scheduled at midnight, the logical_date 2025-01-15 means the run processes data for January 15, even if it actually executes on January 16. Tasks use logical_date to scope their data queries to the correct partition: WHERE event_date = '{{ ds }}' (the ds template variable formats logical_date as YYYY-MM-DD). Using logical_date instead of datetime.now() is essential for correct backfills and replay semantics in event-driven pipelines.

@task
def process(**context):
    ds = context['ds']  # 'YYYY-MM-DD' of the logical date
    df = spark.sql(f"SELECT * FROM events WHERE dt = '{ds}'")

Configure remote logging so task logs persist beyond ephemeral worker containers. In airflow.cfg, set remote_logging=True, remote_base_log_folder=s3://your-bucket/airflow-logs, and remote_log_conn_id=aws_default (an Airflow connection with S3 write permissions). Install apache-airflow-providers-amazon. Logs are uploaded after task completion. The webserver fetches logs from S3 when displaying them in the UI via the same connection. For GCS, use gs:// URI. For Azure, use wasb://. Ensure IAM roles or service accounts grant S3:PutObject and S3:GetObject to the worker and webserver respectively. Log fetch failure in the UI (PermissionError reading S3) is often due to missing webserver IAM permissions even when worker logging succeeds.

# airflow.cfg
[logging]
remote_logging = True
remote_log_conn_id = aws_default
remote_base_log_folder = s3://company-airflow-logs/prod

Prefect is a modern Python-native workflow orchestration tool that emphasizes developer experience and dynamic workflows. Unlike Airflow where DAGs are static (defined at parse time), Prefect flows are plain Python functions with @flow and @task decorators that execute dynamically at runtime. Prefect has no separate scheduler process — flows run via a Prefect deployment on local processes, Docker, or Kubernetes without infrastructure setup for simple cases. State management, retries, caching, and artifacts are first-class Python objects. Prefect Cloud provides a managed orchestration layer. Airflow's strength is its breadth of providers, maturity, and DAG-as-code visibility. Prefect excels for Python-native teams wanting minimal infrastructure overhead.

# Prefect equivalent
from prefect import flow, task
@task(retries=3)
def extract(): ...
@flow
def pipeline():
    data = extract()
    transform(data)

The @task.sensor decorator (Airflow 2.5+) creates a deferrable sensor as a TaskFlow task. The decorated function must return a PokeReturnValue object — if PokeReturnValue(is_done=True, xcom_value=data) is returned, the sensor completes and passes xcom_value downstream. If is_done=False is returned, the task defers and retries after poke_interval seconds, releasing the worker slot between pokes. This replaces subclassing BaseSensorOperator with a lightweight decorator. The deferral mechanism uses the Triggerer component. Use this for custom condition polling (checking a database status column, monitoring an API for job completion) with efficient resource usage.

from airflow.sensors.base import PokeReturnValue
@task.sensor(poke_interval=30, timeout=3600, mode='reschedule')
def wait_for_job(job_id: str) -> PokeReturnValue:
    status = get_job_status(job_id)
    return PokeReturnValue(is_done=(status == 'COMPLETE'), xcom_value=status)