Airflow Production Readiness
When to Use
- Preparing an Airflow deployment for production
- Auditing an existing Airflow installation before a major version upgrade
- Diagnosing scheduler performance issues or slow DAG parsing
- Reviewing new DAG code for production anti-patterns
- Setting up Airflow on Kubernetes with proper resource isolation
Idempotent Task Design
# ✅ Idempotent: INSERT OVERWRITE / UPSERT — safe to re-run
@task
def load_orders(ds: str):
conn = get_trino_connection()
conn.execute(f"""
INSERT OVERWRITE INTO silver.orders
PARTITION (order_date = '{ds}')
SELECT * FROM bronze.orders_raw
WHERE order_date = '{ds}'
""")
# ❌ Not idempotent: appends rows on every run
@task
def load_orders_bad(ds: str):
conn.execute(f"""
INSERT INTO silver.orders
SELECT * FROM bronze.orders_raw WHERE order_date = '{ds}'
""")
No Top-Level Code (DAG Parse Safety)
# ❌ BAD: Variable.get() called at parse time — hammers metadata DB
MY_BUCKET = Variable.get("s3_data_lake_bucket")
with DAG("etl_orders", ...) as dag:
...
# ✅ GOOD: Variable accessed inside task (lazy evaluation)
with DAG("etl_orders", ...) as dag:
@task
def run_etl():
bucket = Variable.get("s3_data_lake_bucket") # called at execution time
...
# ✅ GOOD: Use Jinja template (resolves at task runtime, not parse time)
with DAG("etl_orders", ...) as dag:
BashOperator(
task_id="sync_s3",
bash_command="aws s3 sync /tmp/output {{ var.value.s3_data_lake_bucket }}/output/",
)
# ❌ BAD: expensive import at module level
import pandas as pd # always imported even for unrelated DAGs
import torch # slow, slows every dag parse
# ✅ GOOD: import inside task function
@task
def transform():
import pandas as pd # only imported when task runs
...
Connection and Secret Management
# ✅ Use Airflow Connections for credentials
from airflow.hooks.base import BaseHook
@task
def query_trino():
conn = BaseHook.get_connection("trino_production")
# conn.host, conn.login, conn.password, conn.port
...
# ✅ Use Secret Backends (Vault, AWS SSM)
# airflow.cfg:
# [secrets]
# backend = airflow.providers.amazon.aws.secrets.systems_manager.SystemsManagerParameterStoreBackend
# backend_kwargs = {"connections_prefix": "/airflow/connections", "variables_prefix": "/airflow/variables"}
# ❌ Never hardcode in DAG files:
# DB_PASSWORD = "s3cr3t"
Retry and SLA Configuration
from airflow import DAG
from airflow.utils.dates import days_ago
from datetime import timedelta
with DAG(
dag_id="etl_orders",
start_date=days_ago(1),
schedule="0 2 * * *",
max_active_runs=1, # prevent concurrent runs of same DAG
catchup=False, # don't backfill on deploy
default_args={
"retries": 3,
"retry_delay": timedelta(minutes=5),
"retry_exponential_backoff": True,
"max_retry_delay": timedelta(hours=1),
"on_failure_callback": notify_slack_on_failure,
"sla": timedelta(hours=2), # SLA per task
},
sla_miss_callback=sla_miss_handler, # DAG-level SLA
tags=["orders", "etl"],
) as dag:
...
def sla_miss_handler(dag, task_list, blocking_task_list, slas, blocking_tis):
message = f"SLA missed for DAG {dag.dag_id}\n" \
f"Tasks: {[t.task_id for t in task_list]}"
send_slack_alert(channel="#data-sla-alerts", message=message)
KubernetesExecutor Configuration
# values.yaml (Airflow Helm chart)
executor: KubernetesExecutor
workers:
resources:
requests:
cpu: 500m
memory: 1Gi
limits:
cpu: 2
memory: 4Gi
podAnnotations:
cluster-autoscaler.kubernetes.io/safe-to-evict: "false"
# Pod template for workers
workerPodTemplate: |
apiVersion: v1
kind: Pod
spec:
containers:
- name: base
resources:
requests:
cpu: 500m
memory: 1Gi
serviceAccountName: airflow-worker
tolerations:
- key: spot
operator: Exists
effect: NoSchedule
# Per-task resource override
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
KubernetesPodOperator(
task_id="spark_submit",
image="apache/spark:3.5.0",
cmds=["spark-submit", "--master", "k8s://https://kubernetes.default.svc"],
resources=k8s.V1ResourceRequirements(
requests={"cpu": "2", "memory": "8Gi"},
limits={"cpu": "4", "memory": "16Gi"},
),
is_delete_operator_pod=True,
)
Pool Management
# Limit concurrent database-heavy tasks
# Create pool in Airflow UI or CLI:
# airflow pools set db_intensive_pool 10 "Pool for DB-heavy tasks"
from airflow.operators.python import PythonOperator
load_task = PythonOperator(
task_id="load_to_warehouse",
python_callable=load_fn,
pool="db_intensive_pool", # max 10 concurrent across all DAGs
pool_slots=2, # this task consumes 2 slots
)
Scheduler Configuration (airflow.cfg)
[scheduler]
min_file_process_interval = 30 # re-parse DAG files every 30s (default: 30)
dag_dir_list_interval = 60 # scan dag folder every 60s
max_dagruns_to_create_per_loop = 10
max_tis_per_query = 512
[core]
parallelism = 256 # max concurrent task instances across cluster
dag_concurrency = 32 # max tasks per DAG running concurrently
max_active_tasks_per_dag = 32
max_active_runs_per_dag = 5
[kubernetes_executor]
worker_pods_creation_batch_size = 16 # create up to 16 pods per scheduler loop
delete_worker_pods = True
Metadata DB Maintenance
# Clean old task instances, logs, dag runs (run monthly)
airflow db clean \
--clean-before-timestamp "$(date -d '90 days ago' --utc +%Y-%m-%dT%H:%M:%S)" \
--tables dag_run,task_instance,log,job,xcom \
--yes
# Upgrade DB schema after Airflow version upgrade
airflow db upgrade
# Check metadata DB size (PostgreSQL)
psql $AIRFLOW_DB_CONN -c "
SELECT relname, pg_size_pretty(pg_total_relation_size(relid)) AS size
FROM pg_catalog.pg_statio_user_tables
ORDER BY pg_total_relation_size(relid) DESC
LIMIT 10;
"
Health Check Endpoints
# Airflow health endpoint (liveness probe)
curl http://airflow-webserver:8080/health
# Returns: {"metadatabase": {"status": "healthy"}, "scheduler": {"status": "healthy", "latest_scheduler_heartbeat": "..."}}
# Kubernetes liveness probe
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 30
periodSeconds: 60
failureThreshold: 5
Production Readiness Checklist
[ ] All tasks idempotent (no plain INSERT — use UPSERT/INSERT OVERWRITE)
[ ] No Variable.get() / Connection access at DAG parse time
[ ] No heavy imports at module level
[ ] Retries configured with exponential backoff
[ ] SLA callbacks configured for critical DAGs
[ ] max_active_runs=1 for non-idempotent or resource-heavy DAGs
[ ] catchup=False unless backfill is explicitly needed
[ ] Connections stored in Airflow (not env vars or DAG code)
[ ] Pools configured for DB-intensive and API-rate-limited tasks
[ ] KubernetesExecutor pod resource limits set (no unbounded pods)
[ ] Metadata DB cleaned monthly
[ ] Liveness/readiness probes configured on all Airflow components
[ ] Alert rules: scheduler heartbeat, task failure rate, SLA miss
Anti-Patterns
Variable.get() at DAG parse level — each scheduler parse hits the metadata DB; called hundreds of times per minute at scale; use Jinja templates or access inside tasks.catchup=True with no partition filter — DAG backfills years of data on first deploy; always set catchup=False unless backfill is required.- No
max_active_runs limit — parallel DAG runs compete for DB connections and warehouse resources; set max_active_runs=1 for heavy ETL.
- No pools for rate-limited external APIs — all tasks hit the API simultaneously and trigger rate limits; use pools to cap concurrency.
- Storing secrets in DAG default_args — appears in rendered templates and logs; always use Airflow Connections.
- Not cleaning the metadata DB — the
log and task_instance tables grow unboundedly; schedule monthly airflow db clean.
References
- Airflow best practices:
airflow.apache.org/docs/apache-airflow/stable/best-practices.html
- Airflow Helm chart:
airflow.apache.org/docs/helm-chart/stable/
- KubernetesExecutor:
airflow.apache.org/docs/apache-airflow/stable/core-concepts/executor/kubernetes.html
- Related skills:
[[airflow-dags]], [[dataops-airflow-ha-review]], [[dataops-airflow-observability]]
ivanshamaev