Apache Airflow Deep Dive
Master DAG patterns, dynamic generation, XComs, sensors, and production-grade executor strategies
Beyond Basic DAGs
In Lesson 16, we covered the fundamentals of Apache Airflow, creating basic DAGs, scheduling, and simple task dependencies. But production Airflow deployments require much more: How do you generate 100 DAGs from a configuration file? How do tasks pass data to each other? How do you wait for external events? How do you scale to thousands of concurrent tasks?
This lesson dives deep into advanced Airflow patterns used at companies like Airbnb, Lyft, and Netflix. We'll cover dynamic DAG generation, XComs for task communication, sensors for external triggers, and executor strategies (LocalExecutor, CeleryExecutor, KubernetesExecutor) for scaling to production workloads.
Production DAG Patterns & Best Practices
DAG design patterns that scale from prototype to production. These patterns prevent common pitfalls like task explosion, cascading failures, and maintenance nightmares.
Pattern 1: Idempotent Tasks
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
# ❌ BAD: Non-idempotent (running twice creates duplicates)
def bad_extract(**context):
df = read_from_api()
df.to_parquet(f"s3://bucket/data.parquet", mode='append') # APPEND!
# ✅ GOOD: Idempotent (running twice produces same result)
def good_extract(**context):
execution_date = context['ds'] # YYYY-MM-DD from Airflow
df = read_from_api()
# Use execution date in path (partitioned by date)
output_path = f"s3://bucket/data/date={execution_date}/data.parquet"
df.to_parquet(output_path, mode='overwrite') # OVERWRITE!
print(f"✓ Wrote {len(df)} rows to {output_path}")
return output_path
with DAG(
'idempotent_etl',
start_date=datetime(2024, 1, 1),
schedule='@daily',
catchup=False,
default_args={
'retries': 3, # Safe to retry because idempotent
'retry_delay': timedelta(minutes=5)
}
) as dag:
extract = PythonOperator(
task_id='extract_data',
python_callable=good_extract
)
"""
Why idempotency matters:
- Airflow WILL retry failed tasks
- Manual reruns should produce same result
- Backfills should overwrite, not duplicate
Result: Rerunning this DAG 10 times writes same output file (no duplicates)
"""Best Practice: Always use execution_date in output paths and overwrite mode
Pattern 2: SubDAGs vs TaskGroups
from airflow.decorators import task_group
from airflow.utils.task_group import TaskGroup
# ❌ SubDAGs are DEPRECATED (cause scheduler deadlocks)
# Don't use SubDagOperator!
# ✅ Use TaskGroups for logical grouping (Airflow 2.0+)
with DAG('etl_with_taskgroups', ...) as dag:
start = EmptyOperator(task_id='start')
# TaskGroup 1: Extract from multiple sources
@task_group(group_id='extract_group')
def extract_sources():
@task
def extract_postgres():
return "postgres_data.parquet"
@task
def extract_api():
return "api_data.parquet"
@task
def extract_s3():
return "s3_data.parquet"
# All tasks in group run in parallel
extract_postgres()
extract_api()
extract_s3()
# TaskGroup 2: Transform
@task_group(group_id='transform_group')
def transform_data():
@task
def clean_data():
print("Cleaning data...")
return "cleaned"
@task
def deduplicate():
print("Deduplicating...")
return "deduped"
@task
def enrich():
print("Enriching...")
return "enriched"
# Sequential pipeline within group
clean_data() >> deduplicate() >> enrich()
# TaskGroup 3: Load to multiple targets
@task_group(group_id='load_group')
def load_targets():
@task
def load_redshift():
print("Loading to Redshift...")
@task
def load_elasticsearch():
print("Loading to Elasticsearch...")
@task
def load_s3():
print("Loading to S3...")
# All loads run in parallel
[load_redshift(), load_elasticsearch(), load_s3()]
end = EmptyOperator(task_id='end')
# DAG structure
start >> extract_sources() >> transform_data() >> load_targets() >> end
"""
DAG visualization in Airflow UI:
start
│
▼
┌─────────────────────────────────┐
│ extract_group │
│ ├─ extract_postgres │
│ ├─ extract_api │
│ └─ extract_s3 │
└─────────────────────────────────┘
│
▼
┌─────────────────────────────────┐
│ transform_group │
│ clean_data >> deduplicate │
│ >> enrich │
└─────────────────────────────────┘
│
▼
┌─────────────────────────────────┐
│ load_group │
│ ├─ load_redshift │
│ ├─ load_elasticsearch │
│ └─ load_s3 │
└─────────────────────────────────┘
│
▼
end
Benefits:
✓ Cleaner UI (groups collapse/expand)
✓ Better organization (logical units)
✓ No scheduler issues (unlike SubDAGs)
✓ Easy to reuse across DAGs
"""Result: TaskGroups organize complex DAGs without SubDAG scheduler overhead
Pattern 3: Branching with BranchPythonOperator
from airflow.operators.python import BranchPythonOperator
from airflow.operators.empty import EmptyOperator
def choose_branch(**context):
"""Decide which branch to execute based on business logic"""
logical_date = context['logical_date']
# Example: Different processing on weekends
if logical_date.weekday() >= 5: # Saturday or Sunday
print("Weekend: Running batch processing")
return 'weekend_batch_task'
else:
print("Weekday: Running incremental processing")
return 'weekday_incremental_task'
with DAG('branching_example', ...) as dag:
start = EmptyOperator(task_id='start')
# Branch operator returns task_id of next task to execute
branch = BranchPythonOperator(
task_id='choose_processing_type',
python_callable=choose_branch
)
# Branch 1: Weekday incremental
weekday_task = PythonOperator(
task_id='weekday_incremental_task',
python_callable=lambda: print("Running incremental ETL...")
)
# Branch 2: Weekend batch
weekend_task = PythonOperator(
task_id='weekend_batch_task',
python_callable=lambda: print("Running full batch reload...")
)
# Both branches converge here (use trigger_rule!)
end = EmptyOperator(
task_id='end',
trigger_rule='none_failed_min_one_success' # Run if any upstream succeeded
)
start >> branch >> [weekday_task, weekend_task] >> end
"""
Execution on Monday (weekday):
start → branch → weekday_incremental_task → end
(weekend_batch_task is SKIPPED)
Execution on Saturday (weekend):
start → branch → weekend_batch_task → end
(weekday_incremental_task is SKIPPED)
Trigger Rules:
- all_success: (default) All parents succeeded
- all_failed: All parents failed
- all_done: All parents completed (success or fail)
- one_success: At least one parent succeeded
- one_failed: At least one parent failed
- none_failed: No parents failed (OK if some skipped)
- none_failed_min_one_success: Common for branches!
"""Result: Conditional execution based on date, data size, or external factors
Dynamic DAG Generation
Instead of manually creating 100 similar DAGs, generate them from configuration files. This pattern is used at Airbnb to manage thousands of data pipelines.
import yaml
from pathlib import Path
# ============ STEP 1: Configuration file (config/pipelines.yaml) ============
"""
pipelines:
- name: users_pipeline
source_table: prod.users
destination: s3://warehouse/users/
schedule: '@daily'
- name: orders_pipeline
source_table: prod.orders
destination: s3://warehouse/orders/
schedule: '@hourly'
- name: events_pipeline
source_table: prod.events
destination: s3://warehouse/events/
schedule: '*/15 * * * *' # Every 15 minutes
"""
# ============ STEP 2: DAG generator function ============
def create_etl_dag(dag_id, source_table, destination, schedule):
"""Factory function to create DAG from config"""
@task
def extract(table_name):
print(f"Extracting from {table_name}...")
# Simulate reading from database
return f"extracted_{table_name}.parquet"
@task
def transform(input_file):
print(f"Transforming {input_file}...")
return f"transformed_{input_file}"
@task
def load(input_file, dest):
print(f"Loading {input_file} to {dest}...")
return f"✓ Loaded to {dest}"
with DAG(
dag_id=dag_id,
start_date=datetime(2024, 1, 1),
schedule=schedule,
catchup=False,
tags=['auto-generated', 'etl']
) as dag:
# Create task chain
extracted = extract(source_table)
transformed = transform(extracted)
loaded = load(transformed, destination)
extracted >> transformed >> loaded
return dag
# ============ STEP 3: Generate DAGs from config ============
config_path = Path(__file__).parent / 'config' / 'pipelines.yaml'
with open(config_path) as f:
config = yaml.safe_load(f)
# Dynamically create DAGs (Airflow discovers them!)
for pipeline in config['pipelines']:
dag_id = pipeline['name']
# Create DAG and add to globals() so Airflow can find it
globals()[dag_id] = create_etl_dag(
dag_id=dag_id,
source_table=pipeline['source_table'],
destination=pipeline['destination'],
schedule=pipeline['schedule']
)
print(f"✓ Generated DAG: {dag_id}")
"""
Result: Airflow UI shows 3 DAGs:
• users_pipeline (scheduled @daily)
• orders_pipeline (scheduled @hourly)
• events_pipeline (scheduled */15 * * * *)
Adding new pipeline = adding 5 lines to YAML (no Python code!)
Benefits:
✓ 1 pipeline definition → 1 DAG (DRY principle)
✓ Non-engineers can add pipelines (edit YAML)
✓ Consistent structure across all pipelines
✓ Easy to test (generate DAGs in CI/CD)
✓ Scales to 1000s of pipelines
Warning: Keep config simple! Complex logic should be in Python, not YAML.
"""Result: 3 DAGs generated from YAML config, scalable to thousands of pipelines
Advanced: Generate DAGs per Customer/Tenant
# Multi-tenant SaaS: Generate DAG per customer
import boto3
def get_active_customers():
"""Query database for customers who need ETL"""
# In production: Query PostgreSQL/DynamoDB
return [
{'customer_id': 'acme', 'dataset': 's3://acme-data/'},
{'customer_id': 'globex', 'dataset': 's3://globex-data/'},
{'customer_id': 'initech', 'dataset': 's3://initech-data/'},
]
def create_customer_dag(customer_id, dataset):
"""Create isolated DAG for each customer"""
@task
def process_customer_data(customer, data_path):
print(f"Processing data for {customer} from {data_path}")
# Customer-specific logic here
return f"Processed {customer}"
with DAG(
dag_id=f'customer_etl_{customer_id}', # Unique DAG per customer
start_date=datetime(2024, 1, 1),
schedule='@daily',
catchup=False,
tags=['customer-etl', customer_id]
) as dag:
process_customer_data(customer_id, dataset)
return dag
# Generate DAG for each active customer
for customer in get_active_customers():
globals()[f"customer_etl_{customer['customer_id']}"] = create_customer_dag(
customer_id=customer['customer_id'],
dataset=customer['dataset']
)
"""
Result: Airflow UI shows:
• customer_etl_acme
• customer_etl_globex
• customer_etl_initech
When new customer signs up → DAG automatically appears!
(Airflow scans DAG folder every 30 seconds by default)
Use case: SaaS platforms with per-tenant data pipelines
"""Result: Automatically generate isolated DAG per customer/tenant
XComs: Task-to-Task Communication
XCom (Cross-Communication) lets tasks pass small amounts of data to downstream tasks. Think of it as a message queue between tasks within a DAG run.
XCom Architecture:
Task A Airflow Metadata DB Task B
┌──────────────┐ ┌────────────────────┐ ┌──────────────┐
│ │ │ XCom table │ │ │
│ result = 42 │────────>│ ┌──────────────┐ │────────>│ value = pull │
│ │ push │ │key | value │ │ pull │ print(value) │
│ return result│ │ ├──────────────┤ │ │ │
└──────────────┘ │ │result| "42" │ │ └──────────────┘
│ └──────────────┘ │
└────────────────────┘
⚠️ XCom Limitations:
• Stored in Airflow DB (not for large data!)
• Max size: ~48 KB (SQLite), ~1 GB (PostgreSQL) - but don't!
• Use for metadata, not data
• For large data: Use S3/GCS/filesystem + pass path via XComXComs enable task communication via Airflow's metadata database
Basic XCom Usage
from airflow.decorators import task
@task
def extract_data():
"""Returns value automatically pushed to XCom"""
row_count = 1000
file_path = "s3://bucket/data.parquet"
# Return value is automatically pushed to XCom
return {
'row_count': row_count,
'file_path': file_path,
'status': 'success'
}
@task
def transform_data(metadata):
"""Receives XCom value as argument"""
print(f"Processing {metadata['file_path']}")
print(f"Input rows: {metadata['row_count']}")
# Transform the data
output_path = "s3://bucket/transformed.parquet"
return {
'file_path': output_path,
'row_count': metadata['row_count'] * 2, # Simulated transformation
'status': 'success'
}
@task
def load_data(metadata):
"""Load to warehouse"""
print(f"Loading {metadata['file_path']} ({metadata['row_count']} rows)")
return f"✓ Loaded {metadata['row_count']} rows to warehouse"
with DAG('xcom_example', ...) as dag:
# TaskFlow API automatically handles XCom push/pull
extracted = extract_data()
transformed = transform_data(extracted) # Pass XCom via argument
loaded = load_data(transformed)
extracted >> transformed >> loaded
"""
Execution logs:
[extract_data] INFO - Returning: {'row_count': 1000, 'file_path': 's3://bucket/data.parquet', 'status': 'success'}
[transform_data] INFO - Processing s3://bucket/data.parquet
[transform_data] INFO - Input rows: 1000
[load_data] INFO - Loading s3://bucket/transformed.parquet (2000 rows)
[load_data] INFO - ✓ Loaded 2000 rows to warehouse
XCom values visible in Airflow UI → Admin → XComs
"""Result: Tasks pass metadata (row counts, file paths) via XCom
Advanced: Multiple XCom Values and Custom Keys
from airflow.operators.python import get_current_context
@task
def extract_multiple_sources():
"""Push multiple values to XCom with custom keys"""
context = get_current_context()
ti = context['ti'] # TaskInstance
# Push multiple values with custom keys
ti.xcom_push(key='postgres_rows', value=5000)
ti.xcom_push(key='api_rows', value=3000)
ti.xcom_push(key='s3_rows', value=2000)
# Default return value (key='return_value')
return {'total': 10000}
@task
def validate_counts():
"""Pull specific XCom values"""
context = get_current_context()
ti = context['ti']
# Pull specific keys from upstream task
postgres_rows = ti.xcom_pull(
task_ids='extract_multiple_sources',
key='postgres_rows'
)
api_rows = ti.xcom_pull(
task_ids='extract_multiple_sources',
key='api_rows'
)
s3_rows = ti.xcom_pull(
task_ids='extract_multiple_sources',
key='s3_rows'
)
# Pull default return value
total = ti.xcom_pull(task_ids='extract_multiple_sources')['total']
print(f"PostgreSQL: {postgres_rows:,} rows")
print(f"API: {api_rows:,} rows")
print(f"S3: {s3_rows:,} rows")
print(f"Total: {total:,} rows")
# Validate
actual_total = postgres_rows + api_rows + s3_rows
assert actual_total == total, f"Count mismatch! {actual_total} != {total}"
return "✓ Counts validated"
with DAG('xcom_multiple_values', ...) as dag:
extract_multiple_sources() >> validate_counts()
"""
Output:
[validate_counts] INFO - PostgreSQL: 5,000 rows
[validate_counts] INFO - API: 3,000 rows
[validate_counts] INFO - S3: 2,000 rows
[validate_counts] INFO - Total: 10,000 rows
[validate_counts] INFO - ✓ Counts validated
"""Result: Pass multiple metadata values with custom keys for validation
Best Practice: Large Data via S3 + XCom for Path
import pandas as pd
import boto3
# ❌ BAD: Passing large DataFrame via XCom (will fail!)
@task
def bad_extract():
df = pd.DataFrame({'col': range(1_000_000)}) # 1M rows
return df # TOO BIG for XCom!
# ✅ GOOD: Save to S3, pass path via XCom
@task
def good_extract():
df = pd.DataFrame({'col': range(1_000_000)}) # 1M rows
# Save to S3
s3_path = 's3://bucket/data/extract_output.parquet'
df.to_parquet(s3_path)
# Pass metadata via XCom (small!)
return {
'path': s3_path,
'row_count': len(df),
'columns': list(df.columns),
'size_mb': df.memory_usage(deep=True).sum() / 1024 / 1024
}
@task
def good_transform(metadata):
# Read from S3 using path from XCom
df = pd.read_parquet(metadata['path'])
print(f"Read {metadata['row_count']:,} rows from {metadata['path']}")
print(f"Size: {metadata['size_mb']:.2f} MB")
# Transform
df['col_squared'] = df['col'] ** 2
# Save back to S3
output_path = 's3://bucket/data/transform_output.parquet'
df.to_parquet(output_path)
return {
'path': output_path,
'row_count': len(df),
'size_mb': df.memory_usage(deep=True).sum() / 1024 / 1024
}
with DAG('large_data_pattern', ...) as dag:
extracted = good_extract()
transformed = good_transform(extracted)
"""
Output:
[good_extract] INFO - Saved 1,000,000 rows to s3://bucket/data/extract_output.parquet
[good_extract] INFO - XCom: {'path': '...', 'row_count': 1000000, 'columns': ['col'], 'size_mb': 7.63}
[good_transform] INFO - Read 1,000,000 rows from s3://bucket/data/extract_output.parquet
[good_transform] INFO - Size: 7.63 MB
[good_transform] INFO - Saved to s3://bucket/data/transform_output.parquet
✓ Data in S3 (cheap, durable)
✓ Metadata in XCom (small, fast)
✓ No database bloat
"""Result: Handle millions of rows by storing in S3 and passing paths via XCom
Sensors: Waiting for External Events
Sensors are special operators that wait for an external condition to be met before continuing. They poll periodically until the condition is true or timeout occurs.
from airflow.sensors.filesystem import FileSensor
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.sensors.python import PythonSensor
from datetime import timedelta
# Sensor 1: Wait for file to exist
wait_for_file = FileSensor(
task_id='wait_for_input_file',
filepath='/data/input.csv',
poke_interval=30, # Check every 30 seconds
timeout=3600, # Give up after 1 hour
mode='poke' # 'poke' = block worker, 'reschedule' = release worker
)
# Sensor 2: Wait for S3 object
wait_for_s3 = S3KeySensor(
task_id='wait_for_s3_file',
bucket_name='my-bucket',
bucket_key='data/input.parquet',
aws_conn_id='aws_default',
poke_interval=60,
timeout=7200 # 2 hours
)
# Sensor 3: Custom condition
def check_api_ready():
"""Custom check function"""
import requests
try:
response = requests.get('https://api.example.com/health')
return response.status_code == 200
except:
return False
wait_for_api = PythonSensor(
task_id='wait_for_api',
python_callable=check_api_ready,
poke_interval=10,
timeout=600
)
# Sensor 4: Wait for another DAG to finish
from airflow.sensors.external_task import ExternalTaskSensor
wait_for_upstream_dag = ExternalTaskSensor(
task_id='wait_for_upstream',
external_dag_id='upstream_dag',
external_task_id='final_task', # Wait for specific task
execution_delta=timedelta(hours=1), # Look 1 hour back
poke_interval=60,
timeout=7200
)
with DAG('sensor_example', ...) as dag:
start = EmptyOperator(task_id='start')
# All sensors must complete before processing starts
process = PythonOperator(
task_id='process_data',
python_callable=lambda: print("Processing data...")
)
start >> [wait_for_file, wait_for_s3, wait_for_api] >> process
"""
Sensor modes:
• poke: Task keeps worker occupied (faster, but wastes resources)
• reschedule: Task releases worker and reschedules (efficient for long waits)
Execution timeline:
00:00 - Sensors start poking
00:30 - wait_for_file: File not found (poke again)
01:00 - wait_for_file: File found! ✓
01:00 - wait_for_s3: Checking S3... found ✓
01:00 - wait_for_api: API down (poke again)
01:10 - wait_for_api: API up! ✓
01:10 - All sensors satisfied → process_data starts
Result: process_data only runs when ALL conditions are met
"""Result: DAG waits for files, APIs, and upstream DAGs before processing
Advanced: Smart Reschedule Mode for Long Waits
# For long waits (hours), use reschedule mode to free up workers
wait_for_daily_export = S3KeySensor(
task_id='wait_for_daily_export',
bucket_name='data-exports',
bucket_key='exports/{{ ds }}/data.parquet', # Templated path
poke_interval=300, # Check every 5 minutes
timeout=86400, # Wait up to 24 hours
mode='reschedule', # ✓ Release worker between checks
exponential_backoff=True # ✓ Increase interval over time (5min → 10min → 20min)
)
"""
Reschedule mode timeline:
00:00 - Sensor checks → not found → releases worker, reschedules for 00:05
00:05 - Sensor checks → not found → releases worker, reschedules for 00:15 (backoff)
00:15 - Sensor checks → not found → releases worker, reschedules for 00:35
00:35 - Sensor checks → FOUND! → continues to next task
Worker utilization:
• poke mode: 1 worker × 35 minutes = 35 worker-minutes
• reschedule mode: 1 worker × 4 checks × ~10 sec = 40 worker-seconds
Reschedule mode saves 52× worker capacity!
"""Result: Reschedule mode saves worker capacity for long-running sensors
Executor Strategies: Scaling Airflow to Production
Executors determine how and where tasks run. The right executor choice is critical for production scale, the difference between 10 tasks/second and 1000 tasks/second.
| Executor | Use Case | Parallelism | Pros | Cons |
|---|---|---|---|---|
| SequentialExecutor | Development only | 1 task at a time | Simple, SQLite | Not for production! |
| LocalExecutor | Single machine | 10-100 tasks | Easy setup, low cost | Limited by 1 server |
| CeleryExecutor | Distributed workers | 100-1000s tasks | Scales horizontally | Requires Redis/RabbitMQ |
| KubernetesExecutor | Cloud-native | Unlimited (elastic) | Task isolation, autoscaling | Slower startup (~30s) |
| CeleryKubernetesExecutor | Hybrid (best of both) | Unlimited | Fast for short tasks, elastic for long | Most complex |
LocalExecutor: Single Machine Setup
# airflow.cfg configuration [core] executor = LocalExecutor [database] sql_alchemy_conn = postgresql+psycopg2://airflow:airflow@localhost/airflow [core] parallelism = 32 # Max tasks across all DAGs dag_concurrency = 16 # Max tasks per DAG max_active_runs_per_dag = 1 """ Architecture: ┌─────────────────────────────────────────────────┐ │ Single Server (e.g., EC2 m5.2xlarge) │ │ │ │ ┌──────────────┐ ┌─────────────────────┐ │ │ │ Scheduler │─────>│ LocalExecutor │ │ │ │ (picks tasks│ │ (multiprocessing) │ │ │ │ to run) │ │ │ │ │ └──────────────┘ │ Worker 1 ─ Task A │ │ │ │ Worker 2 ─ Task B │ │ │ ┌──────────────┐ │ Worker 3 ─ Task C │ │ │ │ Web Server │ │ ... │ │ │ │ (Airflow UI)│ │ Worker 32─ Task Z │ │ │ └──────────────┘ └─────────────────────┘ │ │ │ │ ┌──────────────────────────────────────────┐ │ │ │ PostgreSQL │ │ │ │ (Metadata DB) │ │ │ └──────────────────────────────────────────┘ │ └─────────────────────────────────────────────────┘ Capacity: ~32 parallel tasks (limited by cores) Cost: ~$200/month (single m5.2xlarge) Best for: Small-medium teams, <1000 DAG runs/day """
LocalExecutor runs tasks on single machine using multiprocessing
CeleryExecutor: Distributed Workers
# airflow.cfg configuration
[core]
executor = CeleryExecutor
[celery]
broker_url = redis://redis-server:6379/0
result_backend = db+postgresql://airflow:airflow@postgres/airflow
worker_concurrency = 16 # Tasks per worker
"""
Architecture:
┌──────────────────────┐
│ Scheduler │────┐
└──────────────────────┘ │
│
┌──────────────────────┐ │ ┌──────────────────────┐
│ Web Server │ │ │ Redis │
└──────────────────────┘ │ │ (Task Queue) │
└──>│ │
┌──────────────────────┐ └──────┬───────────────┘
│ PostgreSQL │ │
│ (Metadata) │ │
└──────────────────────┘ │
│
┌──────────────────────────────┼──────────────────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Worker Node 1 │ │ Worker Node 2 │ │ Worker Node 3 │
│ ├─ Task A │ │ ├─ Task E │ │ ├─ Task I │
│ ├─ Task B │ │ ├─ Task F │ │ ├─ Task J │
│ ├─ Task C │ │ ├─ Task G │ │ ├─ Task K │
│ └─ Task D │ │ └─ Task H │ │ └─ Task L │
└─────────────────┘ └─────────────────┘ └─────────────────┘
Capacity: 100s-1000s parallel tasks (add more workers!)
Cost: ~$500-5000/month (depends on scale)
Best for: Large teams, >10,000 DAG runs/day
Setup:
# Start worker on each node
airflow celery worker --concurrency 16
# Monitor workers
airflow celery flower # Web UI on port 5555
"""
# Per-task queue assignment (advanced)
high_priority_task = PythonOperator(
task_id='urgent_processing',
python_callable=process_data,
queue='high_priority', # Dedicated workers for urgent tasks
pool='high_priority_pool'
)
low_priority_task = PythonOperator(
task_id='background_job',
python_callable=cleanup,
queue='low_priority' # Separate queue
)
"""
Worker groups:
• 10 workers on 'high_priority' queue (fast servers)
• 50 workers on 'low_priority' queue (spot instances)
Result: Urgent tasks get immediate capacity, background jobs don't block
"""Result: Horizontally scalable with queue-based task distribution
KubernetesExecutor: Cloud-Native Elastic Scaling
# airflow.cfg configuration
[core]
executor = KubernetesExecutor
[kubernetes]
namespace = airflow
worker_container_repository = apache/airflow
worker_container_tag = 2.7.0
"""
Architecture:
┌────────────────────────────────────────────────────────────┐
│ Kubernetes Cluster │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Scheduler │────>│ K8s API │ │
│ │ (creates │ │ Server │ │
│ │ pods) │ └──────┬───────┘ │
│ └──────────────┘ │ │
│ │ │
│ Each task = Dedicated Pod │ │
│ (isolated, ephemeral) │ │
│ │ │
│ ┌──────────────────────┼──────────────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Pod A │ │ Pod B │ │ Pod C │ │
│ │ Task A │ │ Task B │ │ Task C │ │
│ │ 1 CPU │ │ 4 CPU │ │ 16 CPU │ │
│ │ 2 GB │ │ 8 GB │ │ 64 GB │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ (tiny task) (medium) (huge task) │
│ │
└────────────────────────────────────────────────────────────┘
Benefits:
✓ Each task gets dedicated resources (no interference)
✓ Task-specific resource requests (1 CPU or 64 CPUs)
✓ Auto-scaling (Kubernetes HPA scales pods)
✓ Fault isolation (failed task doesn't affect others)
✓ No idle workers (pods terminate after task)
Tradeoffs:
✗ Slower startup (~30 seconds to spin up pod)
✗ Better for long-running tasks (>1 minute)
✗ Not ideal for 1000s of tiny tasks
"""
# Per-task Kubernetes configuration
from kubernetes.client import models as k8s
spark_task = SparkSubmitOperator(
task_id='spark_etl',
application='/path/to/spark_job.py',
executor_config={
'pod_override': k8s.V1Pod(
spec=k8s.V1PodSpec(
containers=[
k8s.V1Container(
name='base',
resources=k8s.V1ResourceRequirements(
requests={'memory': '32Gi', 'cpu': '16'},
limits={'memory': '64Gi', 'cpu': '32'}
)
)
],
node_selector={'workload': 'spark'} # Run on Spark-optimized nodes
)
)
}
)
small_task = PythonOperator(
task_id='quick_check',
python_callable=lambda: print("Hello"),
executor_config={
'pod_override': k8s.V1Pod(
spec=k8s.V1PodSpec(
containers=[
k8s.V1Container(
name='base',
resources=k8s.V1ResourceRequirements(
requests={'memory': '128Mi', 'cpu': '0.1'} # Tiny resources
)
)
]
)
)
}
)
"""
Cost optimization:
• small_task: 0.1 CPU × 30 sec = $0.0001
• spark_task: 16 CPU × 60 min = $2.40
With Celery: Both use same worker size (waste for small_task)
With Kubernetes: Right-sized resources → 50% cost savings!
"""Result: Elastic, isolated execution with task-specific resource allocation
Executor Comparison Matrix
| Metric | LocalExecutor | CeleryExecutor | KubernetesExecutor |
|---|---|---|---|
| Setup complexity | Easy (single machine) | Medium (Redis + workers) | High (Kubernetes cluster) |
| Max parallelism | ~32 tasks (1 machine) | 1000s (add workers) | Unlimited (cluster autoscale) |
| Task startup time | < 1 second | < 1 second | ~30 seconds (pod creation) |
| Resource efficiency | Good (shared resources) | Good (but idle workers) | Excellent (right-sized pods) |
| Fault isolation | No (tasks share process) | Partial (tasks share worker) | Complete (1 task = 1 pod) |
| Cost (monthly) | $100-500 (1 server) | $500-5000 (workers) | $1000-10000 (varies by usage) |
| Best for | Dev/staging, small teams | Production, consistent load | Production, variable load |
| Scaling strategy | Vertical (bigger server) | Horizontal (more workers) | Elastic (autoscaling) |
Bonus: Airflow 3.x Showcase Project
Every pattern covered in this lesson: dynamic DAGs, XComs, sensors, TaskGroups, branching, and CeleryExecutor are demonstrated in a working, Docker Compose project targeting Airflow 3.2.1. Four interconnected DAGs cover real scenarios from ETL pipelines and ML training to cross-DAG orchestration and YAML-driven dynamic generation. It also exercises 3.x-specific features like Asset (event-driven scheduling), the new api-server service, and the separate dag-processor.
The four scenarios:
- weather_etl_pipeline: ETL with
HttpSensor, a customDataQualityOperator, branching, TaskGroups, and an Asset outlet that triggers downstream DAGs on success - ml_training_pipeline: Full TaskFlow API pipeline with asset-driven scheduling: trains a model on the Iris dataset and branches to deploy or retrain based on a configurable accuracy threshold
- city_weather_* (3 DAGs): Dynamic DAG factory: a single Python file reads
config/pipeline_configs.yamland generates one isolated DAG per city viaglobals()- adding a city requires only a new YAML entry - master_orchestrator: Cross-DAG coordination using
TriggerDagRunOperatorto fire all city DAGs in parallel, then trigger the ML pipeline as a fire-and-forget downstream step
# Prerequisites: Docker >= 24 with Compose V2, 4 GB RAM git clone https://gitlab.com/bytecode-solutions/examples/airflow-scheduler cd airflow-scheduler # Start the full stack (PostgreSQL + Redis + MailHog + Airflow services) docker compose up -d --build # Optional: include Flower (Celery monitoring UI at :5555) docker compose --profile flower up -d --build # Airflow UI → http://localhost:8080 (admin / admin) # MailHog → http://localhost:8025 (email notifications) # Trigger Scenario 1 (ETL) docker compose exec airflow-apiserver airflow dags trigger weather_etl_pipeline # Trigger Scenario 3 (Dynamic DAGs) - three city DAGs auto-discovered from YAML docker compose exec airflow-apiserver airflow dags list | grep city_weather # Force the ML pipeline to retrain (raise accuracy threshold) docker compose exec airflow-apiserver airflow variables set ml_accuracy_threshold 0.999 docker compose exec airflow-apiserver airflow dags trigger ml_training_pipeline
Explore the project:gitlab.com/bytecode-solutions/examples/airflow-scheduler
Key Takeaways
- Idempotency: Use execution_date in paths, overwrite mode
- TaskGroups: Organize DAGs, replace deprecated SubDAGs
- Dynamic DAGs: Generate from YAML/database for scalability
- XComs: Pass metadata (not data!) between tasks
- Sensors: Wait for files, APIs, upstream DAGs
- LocalExecutor: Single machine, 32 tasks max
- CeleryExecutor: Distributed, 1000s of tasks
- KubernetesExecutor: Elastic, task-specific resources
Remember: Production Airflow requires thinking beyond basic DAGs. Idempotent tasks prevent duplicate data on retries. Dynamic generation scales to 1000s of pipelines without code duplication. XComs enable task communication but should only pass small metadata, store actual data in S3/GCS. Sensors wait for external events with reschedule mode to avoid blocking workers. Executor choice determines scale: LocalExecutor for small teams, CeleryExecutor for consistent high load, KubernetesExecutor for variable workloads with auto-scaling. Companies like Airbnb run 100,000+ tasks/day using these patterns. Start simple (Local), grow to Celery, adopt Kubernetes when you need elastic scaling and task isolation.