airflow
SKILL.md
Apache Airflow Skill
Master Apache Airflow for workflow orchestration, data pipeline automation, and scheduled task management. This skill covers DAG authoring, operators, sensors, hooks, XComs, variables, connections, and deployment patterns.
When to Use This Skill
USE when:
- Building complex data pipelines with task dependencies
- Orchestrating ETL/ELT workflows
- Scheduling recurring batch jobs
- Managing workflows with retries and error handling
- Coordinating tasks across multiple systems
- Need visibility into workflow execution history
- Requiring audit trails and lineage tracking
- Building ML pipeline orchestration
DON'T USE when:
- Real-time streaming data (use Kafka, Flink)
- Simple cron jobs (use systemd timers, crontab)
- CI/CD pipelines (use GitHub Actions, Jenkins)
- Low-latency requirements (Airflow has scheduler overhead)
- Simple single-task automation (overkill)
- Need visual workflow design for non-developers (use n8n)
Prerequisites
Installation Options
Option 1: pip (Development)
# Create virtual environment
python -m venv airflow-env
source airflow-env/bin/activate
# Set Airflow home
export AIRFLOW_HOME=~/airflow
# Install Airflow with constraints
AIRFLOW_VERSION=2.8.1
PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"
CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"
# Initialize database
airflow db init
# Create admin user
airflow users create \
--username admin \
--firstname Admin \
--lastname User \
--role Admin \
--email admin@example.com \
--password admin
# Start services
airflow webserver --port 8080 &
airflow scheduler &
Option 2: Docker Compose (Recommended)
# Download official docker-compose
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.8.1/docker-compose.yaml'
# Create required directories
mkdir -p ./dags ./logs ./plugins ./config
echo -e "AIRFLOW_UID=$(id -u)" > .env
# Initialize
docker compose up airflow-init
# Start services
docker compose up -d
# Access UI at http://localhost:8080 (airflow/airflow)
Option 3: Kubernetes with Helm
# Add Airflow Helm repo
helm repo add apache-airflow https://airflow.apache.org
helm repo update
# Install Airflow
helm install airflow apache-airflow/airflow \
--namespace airflow \
--create-namespace \
--set executor=KubernetesExecutor
# Get web UI password
kubectl get secret --namespace airflow airflow-webserver-secret -o jsonpath="{.data.webserver-secret-key}" | base64 --decode
Development Setup
# Install development dependencies
pip install apache-airflow[dev,postgres,celery,kubernetes]
# Install testing tools
pip install pytest pytest-airflow
# Install linting
pip install ruff
Core Capabilities
1. Basic DAG Structure
# dags/basic_dag.py
"""
Basic DAG demonstrating core Airflow concepts.
"""
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
# Default arguments for all tasks
default_args = {
'owner': 'data-team',
'depends_on_past': False,
'email': ['alerts@example.com'],
'email_on_failure': True,
'email_on_retry': False,
'retries': 3,
'retry_delay': timedelta(minutes=5),
'retry_exponential_backoff': True,
'max_retry_delay': timedelta(minutes=30),
'execution_timeout': timedelta(hours=2),
}
# DAG definition
with DAG(
dag_id='basic_etl_pipeline',
default_args=default_args,
description='Basic ETL pipeline demonstrating core patterns',
schedule_interval='0 6 * * *', # Daily at 6 AM
start_date=datetime(2026, 1, 1),
catchup=False,
max_active_runs=1,
tags=['etl', 'production'],
doc_md="""
## Basic ETL Pipeline
This DAG demonstrates:
- Task dependencies
- Python and Bash operators
- Error handling with retries
- Task documentation
**Owner**: data-team
**Schedule**: Daily at 6 AM UTC
""",
) as dag:
# Start marker
start = EmptyOperator(
task_id='start',
doc='Pipeline start marker',
)
# Extract task
def extract_data(**context):
"""Extract data from source systems."""
import logging
logger = logging.getLogger(__name__)
# Access execution context
execution_date = context['ds']
logger.info(f"Extracting data for {execution_date}")
# Simulated extraction
data = {
'records': 1000,
'source': 'database',
'execution_date': execution_date,
}
# Return value available via XCom
return data
extract = PythonOperator(
task_id='extract_data',
python_callable=extract_data,
doc='Extract data from source database',
)
# Transform task
def transform_data(**context):
"""Transform extracted data."""
import logging
logger = logging.getLogger(__name__)
# Pull data from previous task via XCom
ti = context['ti']
extracted_data = ti.xcom_pull(task_ids='extract_data')
logger.info(f"Transforming {extracted_data['records']} records")
# Simulated transformation
transformed = {
**extracted_data,
'records_transformed': extracted_data['records'],
'quality_score': 0.95,
}
return transformed
transform = PythonOperator(
task_id='transform_data',
python_callable=transform_data,
doc='Apply transformations to extracted data',
)
# Load task using Bash
load = BashOperator(
task_id='load_data',
bash_command='''
echo "Loading data to warehouse"
echo "Execution date: {{ ds }}"
echo "Previous task output: {{ ti.xcom_pull(task_ids='transform_data') }}"
''',
doc='Load transformed data to data warehouse',
)
# End marker
end = EmptyOperator(
task_id='end',
doc='Pipeline end marker',
trigger_rule='all_success',
)
# Define dependencies
start >> extract >> transform >> load >> end
2. Advanced Operators
# dags/advanced_operators.py
"""
DAG demonstrating advanced operator patterns.
"""
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule
default_args = {
'owner': 'data-team',
'retries': 2,
'retry_delay': timedelta(minutes=5),
}
with DAG(
dag_id='advanced_operators_demo',
default_args=default_args,
schedule_interval=None, # Manual trigger only
start_date=datetime(2026, 1, 1),
catchup=False,
tags=['demo', 'advanced'],
) as dag:
# BranchPythonOperator for conditional logic
def choose_branch(**context):
"""Decide which branch to execute based on data."""
import random
# Simulated decision logic
data_volume = random.randint(0, 1000)
context['ti'].xcom_push(key='data_volume', value=data_volume)
if data_volume > 500:
return 'process_large_dataset'
else:
return 'process_small_dataset'
branch_task = BranchPythonOperator(
task_id='branch_on_data_volume',
python_callable=choose_branch,
)
process_large = PythonOperator(
task_id='process_large_dataset',
python_callable=lambda: print("Processing large dataset with parallel workers"),
)
process_small = PythonOperator(
task_id='process_small_dataset',
python_callable=lambda: print("Processing small dataset directly"),
)
# Join branches - triggered when any upstream succeeds
join = EmptyOperator(
task_id='join_branches',
trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
)
# TaskGroup for logical grouping
from airflow.utils.task_group import TaskGroup
with TaskGroup(group_id='validation_tasks') as validation_group:
validate_schema = PythonOperator(
task_id='validate_schema',
python_callable=lambda: print("Validating data schema"),
)
validate_quality = PythonOperator(
task_id='validate_quality',
python_callable=lambda: print("Validating data quality"),
)
validate_completeness = PythonOperator(
task_id='validate_completeness',
python_callable=lambda: print("Validating data completeness"),
)
# Parallel validation tasks
[validate_schema, validate_quality, validate_completeness]
# Dynamic task mapping (Airflow 2.3+)
def generate_partitions(**context):
"""Generate partition list for dynamic mapping."""
return ['partition_a', 'partition_b', 'partition_c', 'partition_d']
get_partitions = PythonOperator(
task_id='get_partitions',
python_callable=generate_partitions,
)
def process_partition(partition: str):
"""Process a single partition."""
print(f"Processing {partition}")
return f"Processed {partition}"
# Map over partitions dynamically
process_partitions = PythonOperator.partial(
task_id='process_partition',
).expand(op_args=get_partitions.output.map(lambda x: [x]))
# Final aggregation
def aggregate_results(**context):
"""Aggregate results from all partitions."""
ti = context['ti']
results = ti.xcom_pull(task_ids='process_partition', key='return_value')
print(f"Aggregated {len(results)} partition results")
aggregate = PythonOperator(
task_id='aggregate_results',
python_callable=aggregate_results,
)
# Dependencies
branch_task >> [process_large, process_small] >> join
join >> validation_group >> get_partitions >> process_partitions >> aggregate
3. Sensors for Event-Driven Workflows
# dags/sensor_patterns.py
"""
DAG demonstrating sensor patterns for event-driven workflows.
"""
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.sensors.filesystem import FileSensor
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.sensors.python import PythonSensor
from airflow.providers.http.sensors.http import HttpSensor
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
default_args = {
'owner': 'data-team',
'retries': 1,
'retry_delay': timedelta(minutes=1),
}
with DAG(
dag_id='sensor_patterns_demo',
default_args=default_args,
schedule_interval='@daily',
start_date=datetime(2026, 1, 1),
catchup=False,
tags=['sensors', 'event-driven'],
) as dag:
# FileSensor - Wait for file to appear
wait_for_file = FileSensor(
task_id='wait_for_data_file',
filepath='/data/incoming/{{ ds }}/data.csv',
poke_interval=60, # Check every 60 seconds
timeout=3600, # Timeout after 1 hour
mode='poke', # 'poke' or 'reschedule'
soft_fail=False, # Fail task if timeout
)
# S3KeySensor - Wait for S3 object
wait_for_s3 = S3KeySensor(
task_id='wait_for_s3_file',
bucket_name='my-data-bucket',
bucket_key='incoming/{{ ds }}/data.parquet',
aws_conn_id='aws_default',
poke_interval=120,
timeout=7200,
mode='reschedule', # Release worker while waiting
)
# HttpSensor - Wait for API endpoint
wait_for_api = HttpSensor(
task_id='wait_for_api_ready',
http_conn_id='api_connection',
endpoint='/health',
request_params={},
response_check=lambda response: response.json().get('status') == 'ready',
poke_interval=30,
timeout=600,
)
# ExternalTaskSensor - Wait for another DAG
wait_for_upstream_dag = ExternalTaskSensor(
task_id='wait_for_upstream_dag',
external_dag_id='upstream_data_pipeline',
external_task_id='final_task',
execution_delta=timedelta(hours=0), # Same execution date
poke_interval=60,
timeout=3600,
mode='reschedule',
allowed_states=['success'],
failed_states=['failed', 'skipped'],
)
# PythonSensor - Custom condition
def check_data_quality(**context):
"""Custom sensor logic to check data quality."""
import random
# Simulated quality check
quality_score = random.random()
print(f"Quality score: {quality_score}")
return quality_score > 0.8 # Return True when condition met
wait_for_quality = PythonSensor(
task_id='wait_for_data_quality',
python_callable=check_data_quality,
poke_interval=120,
timeout=1800,
mode='poke',
)
# Process after all sensors pass
def process_data(**context):
"""Process data after all conditions are met."""
print("All sensors passed, processing data...")
process = PythonOperator(
task_id='process_data',
python_callable=process_data,
)
# Dependencies - all sensors must pass
[wait_for_file, wait_for_s3, wait_for_api, wait_for_upstream_dag, wait_for_quality] >> process
4. Hooks for External System Integration
# dags/hooks_demo.py
"""
DAG demonstrating hook patterns for external system integration.
"""
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.http.hooks.http import HttpHook
from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
default_args = {
'owner': 'data-team',
'retries': 2,
'retry_delay': timedelta(minutes=5),
}
with DAG(
dag_id='hooks_integration_demo',
default_args=default_args,
schedule_interval='@daily',
start_date=datetime(2026, 1, 1),
catchup=False,
tags=['hooks', 'integration'],
) as dag:
# PostgresHook for database operations
def extract_from_postgres(**context):
"""Extract data from PostgreSQL using hook."""
hook = PostgresHook(postgres_conn_id='postgres_warehouse')
# Execute query and fetch results
sql = """
SELECT id, name, value, created_at
FROM source_table
WHERE created_at >= '{{ ds }}'
AND created_at < '{{ next_ds }}'
"""
# Get connection and execute
connection = hook.get_conn()
cursor = connection.cursor()
cursor.execute(sql)
results = cursor.fetchall()
# Or use pandas
df = hook.get_pandas_df(sql)
print(f"Extracted {len(df)} rows")
# Store record count
context['ti'].xcom_push(key='record_count', value=len(df))
return df.to_dict()
extract_postgres = PythonOperator(
task_id='extract_from_postgres',
python_callable=extract_from_postgres,
)
# S3Hook for file operations
def upload_to_s3(**context):
"""Upload processed data to S3."""
import json
from io import BytesIO
hook = S3Hook(aws_conn_id='aws_default')
# Pull data from previous task
ti = context['ti']
data = ti.xcom_pull(task_ids='extract_from_postgres')
# Convert to JSON and upload
json_data = json.dumps(data)
key = f"processed/{{ ds }}/data.json"
hook.load_string(
string_data=json_data,
key=key,
bucket_name='my-data-bucket',
replace=True,
)
print(f"Uploaded to s3://my-data-bucket/{key}")
return key
upload_s3 = PythonOperator(
task_id='upload_to_s3',
python_callable=upload_to_s3,
)
# HttpHook for API calls
def call_external_api(**context):
"""Make API call using HTTP hook."""
hook = HttpHook(http_conn_id='api_connection', method='POST')
# Prepare payload
payload = {
'execution_date': context['ds'],
'dag_id': context['dag'].dag_id,
'task_id': context['task'].task_id,
}
# Make request
response = hook.run(
endpoint='/api/v1/notify',
data=json.dumps(payload),
headers={'Content-Type': 'application/json'},
)
return response.json()
api_call = PythonOperator(
task_id='call_external_api',
python_callable=call_external_api,
)
# SlackWebhookHook for notifications
def send_slack_notification(**context):
"""Send completion notification to Slack."""
hook = SlackWebhookHook(slack_webhook_conn_id='slack_webhook')
ti = context['ti']
record_count = ti.xcom_pull(task_ids='extract_from_postgres', key='record_count')
message = f"""
:white_check_mark: *Pipeline Completed Successfully*
*DAG*: {context['dag'].dag_id}
*Execution Date*: {context['ds']}
*Records Processed*: {record_count}
"""
hook.send(text=message)
notify_slack = PythonOperator(
task_id='send_slack_notification',
python_callable=send_slack_notification,
)
# Dependencies
extract_postgres >> upload_s3 >> api_call >> notify_slack
5. XCom for Task Communication
# dags/xcom_patterns.py
"""
DAG demonstrating XCom patterns for inter-task communication.
"""
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import XCom
default_args = {
'owner': 'data-team',
'retries': 1,
}
with DAG(
dag_id='xcom_patterns_demo',
default_args=default_args,
schedule_interval=None,
start_date=datetime(2026, 1, 1),
catchup=False,
tags=['xcom', 'communication'],
) as dag:
# Basic XCom push via return value
def produce_data(**context):
"""Produce data - return value auto-pushed to XCom."""
data = {
'records': [
{'id': 1, 'value': 100},
{'id': 2, 'value': 200},
{'id': 3, 'value': 300},
],
'metadata': {
'source': 'api',
'timestamp': str(datetime.now()),
}
}
return data # Automatically pushed to XCom
producer = PythonOperator(
task_id='produce_data',
python_callable=produce_data,
)
# Consume XCom from previous task
def consume_data(**context):
"""Consume data from previous task."""
ti = context['ti']
# Pull return value (default key)
data = ti.xcom_pull(task_ids='produce_data')
print(f"Received {len(data['records'])} records")
# Process and return
total = sum(r['value'] for r in data['records'])
return {'total': total, 'count': len(data['records'])}
consumer = PythonOperator(
task_id='consume_data',
python_callable=consume_data,
)
# Multiple XCom values with custom keys
def produce_multiple(**context):
"""Push multiple XCom values with different keys."""
ti = context['ti']
# Push individual values
ti.xcom_push(key='status', value='success')
ti.xcom_push(key='row_count', value=1000)
ti.xcom_push(key='quality_score', value=0.95)
ti.xcom_push(key='metadata', value={
'source': 'database',
'table': 'transactions',
})
multi_producer = PythonOperator(
task_id='produce_multiple_xcoms',
python_callable=produce_multiple,
)
def consume_multiple(**context):
"""Pull multiple XCom values."""
ti = context['ti']
# Pull specific keys
status = ti.xcom_pull(task_ids='produce_multiple_xcoms', key='status')
row_count = ti.xcom_pull(task_ids='produce_multiple_xcoms', key='row_count')
quality = ti.xcom_pull(task_ids='produce_multiple_xcoms', key='quality_score')
metadata = ti.xcom_pull(task_ids='produce_multiple_xcoms', key='metadata')
print(f"Status: {status}")
print(f"Rows: {row_count}, Quality: {quality}")
print(f"Source: {metadata['source']}")
multi_consumer = PythonOperator(
task_id='consume_multiple_xcoms',
python_callable=consume_multiple,
)
# Cross-DAG XCom (use with caution)
def cross_dag_pull(**context):
"""Pull XCom from another DAG run."""
ti = context['ti']
# Pull from specific DAG
value = ti.xcom_pull(
dag_id='other_dag_id',
task_ids='other_task_id',
key='shared_value',
include_prior_dates=True, # Look at previous runs
)
print(f"Value from other DAG: {value}")
# Template-based XCom access
template_task = PythonOperator(
task_id='template_xcom_access',
python_callable=lambda **ctx: print(ctx['templates_dict']),
templates_dict={
'data': "{{ ti.xcom_pull(task_ids='produce_data') }}",
'status': "{{ ti.xcom_pull(task_ids='produce_multiple_xcoms', key='status') }}",
},
)
# Dependencies
producer >> consumer
multi_producer >> multi_consumer
[consumer, multi_consumer] >> template_task
6. Variables and Connections
# dags/config_management.py
"""
DAG demonstrating Variables and Connections for configuration.
"""
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
from airflow.hooks.base import BaseHook
default_args = {
'owner': 'data-team',
'retries': 1,
}
with DAG(
dag_id='config_management_demo',
default_args=default_args,
schedule_interval=None,
start_date=datetime(2026, 1, 1),
catchup=False,
tags=['config', 'variables'],
) as dag:
# Using Airflow Variables
def use_variables(**context):
"""Access Airflow Variables."""
import json
# Get simple variable
environment = Variable.get('environment', default_var='development')
print(f"Environment: {environment}")
# Get JSON variable (auto-deserialize)
config = Variable.get('pipeline_config', deserialize_json=True)
print(f"Config: {config}")
# Set variable programmatically
Variable.set('last_run', str(datetime.now()))
# Get with default
threshold = Variable.get('quality_threshold', default_var='0.9')
return {
'environment': environment,
'config': config,
'threshold': float(threshold),
}
var_task = PythonOperator(
task_id='use_variables',
python_callable=use_variables,
)
# Using Connections
def use_connections(**context):
"""Access Airflow Connections."""
# Get connection object
conn = BaseHook.get_connection('postgres_warehouse')
print(f"Host: {conn.host}")
print(f"Port: {conn.port}")
print(f"Schema: {conn.schema}")
print(f"Login: {conn.login}")
# Password accessible but don't log it: conn.password
# Get extra fields (JSON)
extra = conn.extra_dejson
print(f"Extra config: {extra}")
# Build connection URI
uri = conn.get_uri()
return {
'host': conn.host,
'schema': conn.schema,
}
conn_task = PythonOperator(
task_id='use_connections',
python_callable=use_connections,
)
# Template-based variable access
def template_vars(**context):
"""Access variables via templates."""
# Variables accessible in templates
print(context['var']['value'])
print(context['var']['json'])
template_task = PythonOperator(
task_id='template_variable_access',
python_callable=template_vars,
op_kwargs={
# Access variable in template
'config': "{{ var.json.pipeline_config }}",
'env': "{{ var.value.environment }}",
},
)
# Environment-specific configuration pattern
def environment_config(**context):
"""Load environment-specific configuration."""
import json
env = Variable.get('environment', default_var='development')
# Load environment-specific config
config_key = f'config_{env}'
config = Variable.get(config_key, deserialize_json=True, default_var={})
# Merge with defaults
defaults = {
'batch_size': 1000,
'timeout': 300,
'retry_count': 3,
}
final_config = {**defaults, **config}
print(f"Final config for {env}: {final_config}")
return final_config
env_config = PythonOperator(
task_id='load_environment_config',
python_callable=environment_config,
)
# Dependencies
[var_task, conn_task] >> template_task >> env_config
7. Error Handling and Callbacks
# dags/error_handling.py
"""
DAG demonstrating error handling and callback patterns.
"""
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.utils.trigger_rule import TriggerRule
def on_success_callback(context):
"""Callback executed on task success."""
print(f"Task {context['task_instance'].task_id} succeeded!")
# Send success notification, update metrics, etc.
def on_failure_callback(context):
"""Callback executed on task failure."""
import logging
logger = logging.getLogger(__name__)
ti = context['task_instance']
dag_id = ti.dag_id
task_id = ti.task_id
execution_date = context['execution_date']
exception = context.get('exception')
error_msg = f"""
Task Failed!
DAG: {dag_id}
Task: {task_id}
Execution Date: {execution_date}
Exception: {exception}
"""
logger.error(error_msg)
# Send alert (Slack, PagerDuty, email, etc.)
# slack_hook.send(text=error_msg)
def on_retry_callback(context):
"""Callback executed on task retry."""
ti = context['task_instance']
print(f"Task {ti.task_id} retrying (attempt {ti.try_number})")
def sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
"""Callback when SLA is missed."""
print(f"SLA missed for tasks: {task_list}")
# Send SLA alert
default_args = {
'owner': 'data-team',
'retries': 3,
'retry_delay': timedelta(minutes=5),
'retry_exponential_backoff': True,
'max_retry_delay': timedelta(minutes=30),
'on_success_callback': on_success_callback,
'on_failure_callback': on_failure_callback,
'on_retry_callback': on_retry_callback,
}
with DAG(
dag_id='error_handling_demo',
default_args=default_args,
schedule_interval='@daily',
start_date=datetime(2026, 1, 1),
catchup=False,
sla_miss_callback=sla_miss_callback,
tags=['error-handling', 'callbacks'],
) as dag:
# Task with potential failure
def potentially_failing_task(**context):
"""Task that might fail."""
import random
if random.random() < 0.3:
raise ValueError("Random failure for demonstration")
return "Success!"
risky_task = PythonOperator(
task_id='risky_task',
python_callable=potentially_failing_task,
sla=timedelta(minutes=30), # SLA deadline
)
# Cleanup task - always runs
cleanup = BashOperator(
task_id='cleanup',
bash_command='echo "Cleaning up resources..."',
trigger_rule=TriggerRule.ALL_DONE, # Run regardless of upstream status
)
# Error handler task - only runs on upstream failure
def handle_error(**context):
"""Handle upstream failures."""
ti = context['ti']
upstream_tasks = ti.get_dagrun().get_task_instances()
failed_tasks = [t for t in upstream_tasks if t.state == 'failed']
if failed_tasks:
print(f"Failed tasks: {[t.task_id for t in failed_tasks]}")
# Implement recovery logic
error_handler = PythonOperator(
task_id='handle_errors',
python_callable=handle_error,
trigger_rule=TriggerRule.ONE_FAILED, # Run if any upstream fails
)
# Success notification - only runs if all succeeded
success_notify = BashOperator(
task_id='success_notification',
bash_command='echo "All tasks completed successfully!"',
trigger_rule=TriggerRule.ALL_SUCCESS,
)
# Dependencies
risky_task >> [cleanup, error_handler, success_notify]
8. Docker and Kubernetes Deployment
# docker-compose.yml - Production-ready Airflow deployment
version: '3.8'
x-airflow-common:
&airflow-common
image: apache/airflow:2.8.1
environment:
&airflow-common-env
AIRFLOW__CORE__EXECUTOR: CeleryExecutor
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@postgres/airflow
AIRFLOW__CELERY__BROKER_URL: redis://:@redis:6379/0
AIRFLOW__CORE__FERNET_KEY: ''
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
AIRFLOW__API__AUTH_BACKENDS: 'airflow.api.auth.backend.basic_auth'
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK: 'true'
# Production settings
AIRFLOW__CORE__PARALLELISM: 32
AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG: 16
AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: 3
AIRFLOW__SCHEDULER__PARSING_PROCESSES: 4
AIRFLOW__CELERY__WORKER_CONCURRENCY: 8
volumes:
- ./dags:/opt/airflow/dags
- ./logs:/opt/airflow/logs
- ./plugins:/opt/airflow/plugins
- ./config:/opt/airflow/config
user: "${AIRFLOW_UID:-50000}:0"
depends_on:
&airflow-common-depends-on
redis:
condition: service_healthy
postgres:
condition: service_healthy
services:
postgres:
image: postgres:15
environment:
POSTGRES_USER: airflow
POSTGRES_PASSWORD: airflow
POSTGRES_DB: airflow
volumes:
- postgres-db-volume:/var/lib/postgresql/data
healthcheck:
test: ["CMD", "pg_isready", "-U", "airflow"]
interval: 10s
retries: 5
start_period: 5s
restart: always
redis:
image: redis:7
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 30s
retries: 50
start_period: 30s
restart: always
airflow-webserver:
<<: *airflow-common
command: webserver
ports:
- "8080:8080"
healthcheck:
test: ["CMD", "curl", "--fail", "http://localhost:8080/health"]
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-scheduler:
<<: *airflow-common
command: scheduler
healthcheck:
test: ["CMD", "curl", "--fail", "http://localhost:8974/health"]
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-worker:
<<: *airflow-common
command: celery worker
healthcheck:
test:
- "CMD-SHELL"
- 'celery --app airflow.providers.celery.executors.celery_executor.app inspect ping -d "celery@$${HOSTNAME}"'
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-triggerer:
<<: *airflow-common
command: triggerer
healthcheck:
test: ["CMD-SHELL", 'airflow jobs check --job-type TriggererJob --hostname "$${HOSTNAME}"']
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-init:
<<: *airflow-common
entrypoint: /bin/bash
command:
- -c
- |
airflow db init
airflow users create \
--username admin \
--password admin \
--firstname Admin \
--lastname User \
--role Admin \
--email admin@example.com
environment:
<<: *airflow-common-env
_AIRFLOW_DB_MIGRATE: 'true'
_AIRFLOW_WWW_USER_CREATE: 'true'
user: "0:0"
volumes:
postgres-db-volume:
# kubernetes/values.yaml - Helm chart values for Kubernetes deployment
executor: KubernetesExecutor
# Airflow configuration
config:
core:
dags_are_paused_at_creation: 'true'
load_examples: 'false'
parallelism: 32
max_active_tasks_per_dag: 16
scheduler:
parsing_processes: 4
kubernetes:
delete_worker_pods: 'true'
delete_worker_pods_on_failure: 'false'
# Web server
webserver:
replicas: 2
resources:
requests:
memory: "1Gi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "1000m"
# Scheduler
scheduler:
replicas: 2
resources:
requests:
memory: "2Gi"
cpu: "1000m"
limits:
memory: "4Gi"
cpu: "2000m"
# Worker pod template
workers:
resources:
requests:
memory: "1Gi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "1000m"
# DAGs configuration
dags:
persistence:
enabled: true
size: 10Gi
gitSync:
enabled: true
repo: https://github.com/org/airflow-dags.git
branch: main
wait: 60
# Logs
logs:
persistence:
enabled: true
size: 50Gi
# Database
postgresql:
enabled: true
persistence:
enabled: true
size: 20Gi
# Redis (for Celery if using)
redis:
enabled: false
# Ingress
ingress:
enabled: true
web:
annotations:
kubernetes.io/ingress.class: nginx
cert-manager.io/cluster-issuer: letsencrypt-prod
hosts:
- name: airflow.example.com
tls:
enabled: true
secretName: airflow-tls
Integration Examples
Integration with AWS Services
# dags/aws_integration.py
"""
DAG integrating with AWS services.
"""
from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.amazon.aws.operators.s3 import S3CreateBucketOperator
from airflow.providers.amazon.aws.transfers.local_to_s3 import LocalFilesystemToS3Operator
from airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator
from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
from airflow.providers.amazon.aws.operators.athena import AthenaOperator
default_args = {
'owner': 'data-team',
'retries': 2,
}
with DAG(
dag_id='aws_integration_pipeline',
default_args=default_args,
schedule_interval='@daily',
start_date=datetime(2026, 1, 1),
catchup=False,
tags=['aws', 'integration'],
) as dag:
# Upload to S3
upload_to_s3 = LocalFilesystemToS3Operator(
task_id='upload_to_s3',
filename='/data/output/{{ ds }}/data.parquet',
dest_key='raw/{{ ds }}/data.parquet',
dest_bucket='my-data-lake',
aws_conn_id='aws_default',
replace=True,
)
# Run Glue ETL job
run_glue_job = GlueJobOperator(
task_id='run_glue_etl',
job_name='my-etl-job',
script_args={
'--input_path': 's3://my-data-lake/raw/{{ ds }}/',
'--output_path': 's3://my-data-lake/processed/{{ ds }}/',
},
aws_conn_id='aws_default',
wait_for_completion=True,
)
# Query with Athena
run_athena_query = AthenaOperator(
task_id='run_athena_analysis',
query="""
SELECT date, COUNT(*) as count, SUM(value) as total
FROM processed_data
WHERE partition_date = '{{ ds }}'
GROUP BY date
""",
database='analytics',
output_location='s3://my-data-lake/athena-results/',
aws_conn_id='aws_default',
)
# Load to Redshift
load_to_redshift = S3ToRedshiftOperator(
task_id='load_to_redshift',
schema='public',
table='fact_daily_metrics',
s3_bucket='my-data-lake',
s3_key='processed/{{ ds }}/',
redshift_conn_id='redshift_warehouse',
aws_conn_id='aws_default',
copy_options=['FORMAT AS PARQUET'],
)
upload_to_s3 >> run_glue_job >> run_athena_query >> load_to_redshift
Best Practices
1. DAG Design Principles
# Use meaningful DAG and task IDs
dag_id='sales_daily_etl_pipeline' # Good
dag_id='dag1' # Bad
# Set appropriate concurrency limits
max_active_runs=1 # For data pipelines with dependencies
max_active_tasks_per_dag=16 # Limit resource usage
# Use tags for organization
tags=['production', 'etl', 'sales']
# Always set catchup=False unless backfill needed
catchup=False
# Use execution_timeout to prevent stuck tasks
execution_timeout=timedelta(hours=2)
2. Task Best Practices
# Keep tasks atomic and idempotent
def process_partition(partition_date: str):
"""Idempotent: can be safely re-run."""
# Delete existing data for this partition
delete_partition(partition_date)
# Process and insert new data
insert_data(partition_date)
# Use retries with exponential backoff
default_args = {
'retries': 3,
'retry_delay': timedelta(minutes=5),
'retry_exponential_backoff': True,
}
# Avoid heavy processing in sensors
# Bad: sensor does complex computation
# Good: sensor checks simple condition, processing in separate task
3. Configuration Management
# Use Variables for configuration, not hardcoded values
batch_size = Variable.get('batch_size', default_var=1000)
# Use Connections for credentials
conn = BaseHook.get_connection('my_database')
# Environment-specific configuration
env = Variable.get('environment')
config = Variable.get(f'config_{env}', deserialize_json=True)
4. Testing DAGs
# tests/test_dags.py
import pytest
from airflow.models import DagBag
def test_dag_loads():
"""Test that DAGs load without errors."""
dagbag = DagBag()
assert len(dagbag.import_errors) == 0
def test_dag_structure():
"""Test DAG has expected structure."""
dagbag = DagBag()
dag = dagbag.get_dag('my_pipeline')
assert dag is not None
assert len(dag.tasks) == 5
assert dag.schedule_interval == '@daily'
Troubleshooting
Common Issues
Issue: DAG not appearing in UI
# Check for import errors
airflow dags list-import-errors
# Validate DAG file
python dags/my_dag.py
# Check scheduler logs
docker logs airflow-scheduler-1 | grep -i error
Issue: Tasks stuck in queued state
# Check worker status
airflow celery status
# Verify executor configuration
airflow config get-value core executor
# Check for resource constraints
kubectl top pods -n airflow
Issue: XCom size limits
# Use external storage for large data
def store_large_result(**context):
# Store in S3 instead of XCom
s3_hook.load_string(large_data, key='results/data.json', bucket='my-bucket')
return 's3://my-bucket/results/data.json' # Return reference only
Issue: Scheduler performance
# Tune scheduler settings
AIRFLOW__SCHEDULER__PARSING_PROCESSES: 4
AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL: 30
AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 60
Debugging Tips
# Add detailed logging
import logging
logger = logging.getLogger(__name__)
def my_task(**context):
logger.info(f"Starting task with context: {context}")
# ... task logic
logger.debug(f"Intermediate result: {result}")
# Test specific task
airflow tasks test my_dag my_task 2026-01-15
# Clear task state for re-run
airflow tasks clear my_dag -t my_task -s 2026-01-15 -e 2026-01-15
# Trigger DAG run
airflow dags trigger my_dag --conf '{"key": "value"}'
Version History
| Version | Date | Changes |
|---|---|---|
| 1.0.0 | 2026-01-17 | Initial release with comprehensive workflow patterns |
Resources
- Apache Airflow Documentation
- Airflow Best Practices
- Airflow Helm Chart
- Astronomer Guides
- Airflow Providers
This skill provides production-ready patterns for Apache Airflow workflow orchestration, tested across enterprise data pipelines.
Weekly Installs
7
Repository
vamseeachanta/workspace-hubFirst Seen
Jan 24, 2026
Security Audits
Installed on
claude-code6
trae5
antigravity5
codex5
windsurf5
gemini-cli5