Apache Airflow Cheat Sheet

Overview

Apache Airflow is an open-source platform for programmatically authoring, scheduling, and monitoring workflows. Originally developed at Airbnb in 2014, it became an Apache top-level project and is now the industry standard for data pipeline orchestration. Airflow represents workflows as Directed Acyclic Graphs (DAGs) defined in Python, giving engineers full programmatic control over pipeline logic, dependencies, and scheduling.

Airflow’s architecture consists of a scheduler that triggers tasks, an executor that runs them (local, Celery, or Kubernetes), a metadata database that stores state, and a web UI for monitoring. It supports a rich ecosystem of provider packages for connecting to cloud services, databases, APIs, and more. With features like dynamic DAG generation, XCom for inter-task communication, and a powerful plugin system, Airflow handles everything from simple cron-like schedules to complex multi-stage data pipelines with conditional branching and retry logic.

Installation

Install via pip

# Set Airflow home directory
export AIRFLOW_HOME=~/airflow

# Install Apache Airflow with constraints
AIRFLOW_VERSION=2.9.0
PYTHON_VERSION="$(python3 --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 the database
airflow db init

# Create an admin user
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email admin@example.com \
    --password admin

# Start the webserver and scheduler
airflow webserver --port 8080 &
airflow scheduler &

Install via Docker Compose

# Download the official docker-compose file
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.9.0/docker-compose.yaml'

# Create required directories
mkdir -p ./dags ./logs ./plugins ./config
echo -e "AIRFLOW_UID=$(id -u)" > .env

# Initialize Airflow
docker compose up airflow-init

# Start all services
docker compose up -d

Install with Helm on Kubernetes

# Add the Airflow Helm chart 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

Core Concepts

Basic DAG Definition

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['alerts@example.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'execution_timeout': timedelta(hours=1),
}

with DAG(
    dag_id='etl_pipeline',
    default_args=default_args,
    description='Daily ETL pipeline',
    schedule='0 6 * * *',  # Daily at 6 AM
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['etl', 'production'],
    max_active_runs=1,
) as dag:

    extract = BashOperator(
        task_id='extract_data',
        bash_command='python /scripts/extract.py',
    )

    transform = PythonOperator(
        task_id='transform_data',
        python_callable=lambda: print("Transforming..."),
    )

    load = BashOperator(
        task_id='load_data',
        bash_command='python /scripts/load.py',
    )

    extract >> transform >> load

TaskFlow API (Airflow 2.x)

from airflow.decorators import dag, task
from datetime import datetime

@dag(
    schedule='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['taskflow'],
)
def etl_taskflow():

    @task()
    def extract():
        return {"data": [1, 2, 3]}

    @task()
    def transform(raw_data: dict):
        return {"transformed": [x * 2 for x in raw_data["data"]]}

    @task()
    def load(transformed_data: dict):
        print(f"Loading: {transformed_data}")

    raw = extract()
    transformed = transform(raw)
    load(transformed)

etl_taskflow()

CLI Commands

Command	Description
airflow dags list	List all registered DAGs
airflow dags trigger <dag_id>	Trigger a DAG run
airflow dags pause <dag_id>	Pause a DAG
airflow dags unpause <dag_id>	Unpause a DAG
airflow dags test <dag_id> <date>	Test a DAG without writing to DB
airflow tasks list <dag_id>	List tasks in a DAG
airflow tasks test <dag_id> <task_id> <date>	Test a single task
airflow tasks run <dag_id> <task_id> <date>	Run a single task
airflow dags backfill <dag_id> -s <start> -e <end>	Backfill DAG runs
airflow db check	Check database connectivity
airflow db upgrade	Upgrade the metadata database schema
airflow db reset	Reset the metadata database
airflow connections list	List all configured connections
airflow connections add <conn_id> --conn-uri <uri>	Add a connection
airflow variables set <key> <value>	Set an Airflow variable
airflow variables get <key>	Get an Airflow variable
airflow pools list	List all resource pools
airflow pools set <name> <slots> <description>	Create/update a pool
airflow config list	List current configuration
airflow info	Show Airflow environment info

Configuration

airflow.cfg Key Settings

[core]
dags_folder = /opt/airflow/dags
executor = CeleryExecutor
load_examples = False
default_timezone = UTC
max_active_runs_per_dag = 16
parallelism = 32
dag_file_processor_timeout = 150

[webserver]
web_server_port = 8080
rbac = True
expose_config = False

[scheduler]
min_file_process_interval = 30
dag_dir_list_interval = 300
parsing_processes = 4

[celery]
broker_url = redis://redis:6379/0
result_backend = db+postgresql://airflow:airflow@postgres/airflow

[logging]
base_log_folder = /opt/airflow/logs
remote_logging = True
remote_log_conn_id = aws_default
remote_base_log_folder = s3://my-bucket/airflow/logs

Connection Configuration

# Add connections via CLI
airflow connections add 'postgres_default' \
    --conn-type 'postgres' \
    --conn-host 'db.example.com' \
    --conn-port 5432 \
    --conn-login 'user' \
    --conn-password 'pass' \
    --conn-schema 'mydb'

airflow connections add 'aws_default' \
    --conn-type 'aws' \
    --conn-extra '{"region_name": "us-east-1"}'

airflow connections add 'gcp_default' \
    --conn-type 'google_cloud_platform' \
    --conn-extra '{"key_path": "/path/to/keyfile.json", "project": "my-project"}'

Advanced Usage

Dynamic DAG Generation

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

databases = ['users', 'orders', 'products', 'inventory']

for db in databases:
    dag_id = f'sync_{db}'

    with DAG(
        dag_id=dag_id,
        schedule='@hourly',
        start_date=datetime(2024, 1, 1),
        catchup=False,
    ) as dag:
        sync = BashOperator(
            task_id=f'sync_{db}_task',
            bash_command=f'python /scripts/sync.py --database {db}',
        )

    globals()[dag_id] = dag

Branching and Conditional Logic

from airflow.operators.python import BranchPythonOperator
from airflow.operators.empty import EmptyOperator

def choose_branch(**context):
    execution_date = context['logical_date']
    if execution_date.weekday() < 5:
        return 'weekday_processing'
    return 'weekend_processing'

branch = BranchPythonOperator(
    task_id='branch_check',
    python_callable=choose_branch,
)

weekday = EmptyOperator(task_id='weekday_processing')
weekend = EmptyOperator(task_id='weekend_processing')
join = EmptyOperator(task_id='join', trigger_rule='none_failed_min_one_success')

branch >> [weekday, weekend] >> join

Custom Sensors and Operators

from airflow.sensors.base import BaseSensorOperator
from airflow.utils.decorators import apply_defaults

class FileSensor(BaseSensorOperator):
    @apply_defaults
    def __init__(self, filepath, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.filepath = filepath

    def poke(self, context):
        import os
        return os.path.exists(self.filepath)

wait_for_file = FileSensor(
    task_id='wait_for_file',
    filepath='/data/incoming/report.csv',
    poke_interval=60,
    timeout=3600,
    mode='reschedule',
)

Task Groups

from airflow.utils.task_group import TaskGroup

with DAG('grouped_dag', schedule='@daily', start_date=datetime(2024, 1, 1)) as dag:
    with TaskGroup('extract_group') as extract:
        extract_users = BashOperator(task_id='users', bash_command='echo users')
        extract_orders = BashOperator(task_id='orders', bash_command='echo orders')

    with TaskGroup('transform_group') as transform:
        transform_data = BashOperator(task_id='transform', bash_command='echo transform')

    extract >> transform

Troubleshooting

Issue	Solution
DAG not appearing in UI	Check for Python syntax errors: python /path/to/dag.py. Ensure file is in dags_folder
Tasks stuck in queued state	Check executor capacity, pool slots, and worker availability. Verify parallelism settings
Database connection errors	Verify sql_alchemy_conn in config. Run airflow db check
Scheduler not picking up DAGs	Increase min_file_process_interval. Check scheduler logs
XCom too large	Use external storage (S3/GCS) for large data. XCom is for metadata only
Import errors in DAGs	Install missing providers: pip install apache-airflow-providers-*
Webserver 502 errors	Increase web_server_worker_timeout. Check memory usage
Tasks timing out	Adjust execution_timeout in default_args or per-task
Zombie tasks detected	The scheduler auto-cleans these. Check worker health if persistent
Permission denied on logs	Ensure AIRFLOW_UID matches container user. Fix with chmod -R 777 ./logs