Apache Airflow Tutorial: Workflow Orchestration at Scale (2026)
I have spent years building data pipelines with Apache Airflow, and it remains the most battle-tested workflow orchestrator in the data engineering ecosystem. After migrating from cron scripts and custom orchestrators to Airflow, I realized how much operational overhead disappears when you have a centralized, auditable, and retryable system managing your dependencies.
This tutorial covers Airflow's architecture, DAG authoring, operators, sensors, task relationships, executors, and production best practices for building reliable data pipelines.
Airflow Architecture and Core Concepts
Airflow consists of a scheduler, a web server, a metadata database, and one or more executors. The scheduler解析s DAG files, resolves task dependencies, and submits tasks to the executor. The metadata database (PostgreSQL in production) tracks DAG runs, task states, and historical execution data. Workers pick up tasks from the executor queue and execute them.
DAGs (Directed Acyclic Graphs) define the workflow structure. Each DAG contains tasks connected by dependencies. The scheduler resolves the execution order using a topological sort. A task instance represents a single execution of a task in a specific DAG run.
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
def extract_data():
print('Extracting data...')
def transform_data():
print('Transforming data...')
def load_data():
print('Loading data...')
with DAG(
dag_id='etl_pipeline',
start_date=datetime(2026, 1, 1),
schedule='@daily',
catchup=False
) as dag:
extract = PythonOperator(task_id='extract', python_callable=extract_data)
transform = PythonOperator(task_id='transform', python_callable=transform_data)
load = PythonOperator(task_id='load', python_callable=load_data)
extract >> transform >> loadOperators, Sensors, and Hooks
Operators are the building blocks of Airflow tasks. Each operator encapsulates a specific action: running a Bash command, querying a database, submitting a Spark job, or calling an API. Operators are declarative — you define what should happen, and Airflow handles execution, retries, and logging.
Sensors are a specialized operator that waits for an external condition to be true before completing. For example, a FileSensor waits for a file to appear in HDFS; an HttpSensor waits for an endpoint to return 200. Sensors can run in poke mode (polling in a loop) or reschedule mode (releasing the worker slot between pokes).
from airflow.operators.bash import BashOperator
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.postgres.operators.postgres import PostgresOperator
# Bash operator
task1 = BashOperator(
task_id='run_script',
bash_command='python /opt/scripts/process.py',
retries=3,
retry_delay=timedelta(minutes=5)
)
# S3 sensor — waits for file
task2 = S3KeySensor(
task_id='wait_for_file',
bucket_name='my-bucket',
bucket_key='data/input.parquet',
timeout=3600,
poke_interval=60
)
# Postgres operator
task3 = PostgresOperator(
task_id='update_table',
postgres_conn_id='my_db',
sql='UPDATE metrics SET updated_at = NOW() WHERE date = %s',
parameters=('2026-01-01',)
)TaskFlow API and XComs
TaskFlow API (introduced in Airflow 2.0) simplifies passing data between tasks using Python decorators and XComs. XComs (cross-communication) let tasks push and pull small amounts of data. The @task decorator automatically handles pushing the return value to XCom, and downstream tasks can pull it using the task instance.
XComs are designed for small metadata, not large datasets. For passing large data, use external storage (S3, GCS) and pass references via XCom. The default XCom backend stores data in the metadata database, which has practical size limits.
from airflow.decorators import task, dag
from datetime import datetime
@dag(schedule='@daily', start_date=datetime(2026, 1, 1), catchup=False)
def data_pipeline():
@task
def extract():
data = [1, 2, 3, 4, 5]
return data
@task
def transform(raw_data):
return [x * 2 for x in raw_data]
@task
def load(processed_data):
print(f'Loading {len(processed_data)} records')
raw = extract()
transformed = transform(raw)
load(transformed)
data_pipeline()Executors and Parallelism
Executors determine how tasks are distributed across workers. The SequentialExecutor runs tasks one at a time (for testing only). The LocalExecutor runs tasks in parallel using multiprocessing on a single machine. The CeleryExecutor distributes tasks across a pool of worker nodes using a message broker. The KubernetesExecutor spins up a pod per task.
For production, CeleryExecutor or KubernetesExecutor is recommended. CeleryExecutor is simpler to manage with a fixed pool of workers. KubernetesExecutor offers better resource isolation and dynamic scaling but adds operational complexity.
# airflow.cfg
[core]
executor = CeleryExecutor
parallelism = 32
max_active_tasks_per_dag = 16
[celery]
broker_url = redis://redis:6379/0
result_backend = db+postgresql://airflow:airflow@postgres/airflow
# KubernetesExecutor config
[core]
executor = KubernetesExecutor
[kubernetes]
namespace = airflow
worker_container_repository = my-registry/airflow-worker
worker_container_tag = latest
dags_in_image = TrueProduction Best Practices
Do not store state in tasks — Airflow tasks should be idempotent. If a task fails and retries, it should produce the same result. Use task parameters and XComs to pass data, not global variables or local files that may not persist across retries.
Use connection pools and connection IDs consistently. Store credentials in Airflow Connections (backed by the metadata database or a secrets backend like HashiCorp Vault). Never hardcode passwords in DAG files. Use Variable.get() sparingly since it queries the database on every call.
from airflow.models import Connection
from airflow.hooks.base import BaseHook
# Better: use connection IDs
conn = BaseHook.get_connection('my_database')
print(f'Host: {conn.host}, Port: {conn.port}')
# Use secrets backend
# In airflow.cfg:
# [secrets]
# backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
# backend_kwargs = {"connections_path": "airflow/connections"}
# Test connections via CLI
# airflow connections get my_databaseMonitoring, Alerting, and SLAs
Airflow provides built-in monitoring through the web UI, which shows DAG runs, task states, and execution timelines. For production, integrate with external monitoring systems. Use SLA misses to alert when a DAG does not complete within its expected timeframe. Configure callbacks for task failure, retries, and sla_miss events.
Airflow emits metrics via StatsD that can be consumed by Prometheus. Monitor scheduler lag, pool utilization, and executor queue depth. Set up alerts for tasks that exceed expected durations or fail repeatedly.
from airflow.callbacks import callback
from airflow.models import SLAMiss
# SLA configuration
dag = DAG(
dag_id='daily_etl',
sla_miss_callback=notify_sla_failure,
default_args={
'sla': timedelta(hours=2),
'on_failure_callback': notify_task_failure,
'retries': 2,
'retry_delay': timedelta(minutes=10)
}
)
# Custom alerting
def notify_task_failure(context):
send_alert(
task=context['task'],
dag=context['dag'],
exception=context['exception']
)
# StatsD metrics are emitted automatically
# scheduler_lag, pool_usage, task_success/failure countsFrequently Asked Questions
What is the difference between Airflow 1.x and 2.x?
Airflow 2.x introduced the TaskFlow API, dynamic task mapping, AIP-39 (timetables), and improved scheduler performance. The CeleryExecutor was decoupled into the apache-airflow-providers-celery package.
How does Airflow handle task retries?
When a task fails, Airflow retries it up to the configured number of retries with a specified delay. Retries are recorded in the metadata database, and the task transitions through states: queued, running, up_for_retry, failed, or success.
Can I use Airflow for streaming workflows?
Airflow is designed for batch workflows. For streaming, consider Apache Flink or Spark Structured Streaming. However, Airflow can orchestrate batch micro-batches that simulate near-real-time processing.
What is dynamic task mapping?
Dynamic task mapping (Airflow 2.3+) allows tasks to be created at runtime based on the output of upstream tasks. This enables fan-out patterns where a single task generates multiple downstream tasks dynamically.
Originally published on Ayodhyyya. Last updated June 1, 2026.