Apache Airflow¶

Apache Airflow is a platform for programmatically creating, scheduling, and monitoring workflows. This sheet covers DAGs, operators, scheduling, sensors, connections, XComs, and task patterns.

Table of Contents¶

Installation / Setup
Core Syntax
Common Operations
Practical Examples
Common Patterns
Best Practices
Common Mistakes
Troubleshooting
Related Cheat Sheets
Official Documentation

Installation / Setup¶

# Install Airflow
pip install apache-airflow

# 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 webserver
airflow webserver --port 8080

# Start scheduler
airflow scheduler

Core Syntax¶

DAG Definition¶

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

# Default arguments
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG definition
with DAG(
    dag_id='my_pipeline',
    default_args=default_args,
    description='My data pipeline',
    schedule_interval='@daily',
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['example'],
) as dag:

    task1 = PythonOperator(
        task_id='extract',
        python_callable=extract_data,
    )

    task2 = PythonOperator(
        task_id='transform',
        python_callable=transform_data,
    )

    task1 >> task2

Operators¶

from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from airflow.providers.amazon.aws.operators.s3 import S3CreateObjectOperator
from airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator

# PythonOperator
def my_function(**context):
    print("Hello from Airflow")
    return "result"

task = PythonOperator(
    task_id='my_task',
    python_callable=my_function,
)

# BashOperator
task = BashOperator(
    task_id='run_script',
    bash_command='python /path/to/script.py',
)

# EmptyOperator (placeholder)
task = EmptyOperator(task_id='noop')

Task Dependencies¶

# Sequential
task1 >> task2 >> task3

# Parallel
task1 >> [task2, task3]

# Complex
task1 >> task2
task1 >> task3
task2 >> task4
task3 >> task4

XCom (Cross-Communication)¶

# Push value
def push_value(**context):
    context['ti'].xcom_push(key='my_key', value='my_value')

# Pull value
def pull_value(**context):
    value = context['ti'].xcom_pull(key='my_key', task_ids='push_task')
    print(f"Got: {value}")

# Using return value (automatically pushed)
def return_value():
    return 42

task = PythonOperator(
    task_id='return_task',
    python_callable=return_value,
)

# Pull in another task
task2 = PythonOperator(
    task_id='pull_task',
    python_callable=lambda **ctx: ctx['ti'].xcom_pull(task_ids='return_task'),
)

Connections¶

from airflow.hooks.base import BaseHook

# Get connection
conn = BaseHook.get_connection('my_conn')
print(conn.host)
print(conn.login)
print(conn.password)

# Using in operators
from airflow.providers.amazon.aws.hooks.s3 import S3Hook

hook = S3Hook(aws_conn_id='aws_default')
bucket = hook.get_bucket('my-bucket')

Sensors¶

from airflow.sensors.filesystem import FileSensor
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.sensors.http_sensor import HttpSensor

# Wait for file
file_sensor = FileSensor(
    task_id='wait_for_file',
    filepath='/data/input/file.csv',
    poke_interval=60,
)

# Wait for HTTP endpoint
http_sensor = HttpSensor(
    task_id='wait_for_api',
    http_conn_id='my_api',
    endpoint='/health',
)

# Wait for another DAG
ext_sensor = ExternalTaskSensor(
    task_id='wait_for_upstream',
    external_dag_id='upstream_dag',
    external_task_id='final_task',
)

Templates and Jinja¶

# Use Jinja templates in operators
task = BashOperator(
    task_id='print_date',
    bash_command='echo {{ ds }}',  # execution date
)

# Templates in PythonOperator
def my_func(**context):
    ds = context['ds']  # execution date
    execution_date = context['execution_date']
    print(f"Running for {ds}")

Common Operations¶

Schedule Intervals¶

# Standard intervals
schedule_interval='@daily'      # once a day
schedule_interval='@hourly'     # once an hour
schedule_interval='@weekly'     # once a week
schedule_interval='@monthly'    # once a month

# Cron expressions
schedule_interval='0 6 * * *'   # daily at 6am
schedule_interval='*/15 * * * *' # every 15 minutes
schedule_interval='0 8 * * 1-5' # weekdays at 8am

Context Variables¶

# Available in all tasks
context = {
    'ds': '2026-01-01',              # execution date
    'ds_nodash': '20260101',         # execution date no dashes
    'execution_date': datetime,      # execution date object
    'next_execution_date': datetime, # next scheduled date
    'prev_execution_date': datetime, # previous execution date
    'dag': DAG,                      # DAG object
    'task': Task,                    # task object
    'ti': TaskInstance,              # task instance
    'run_id': 'scheduled__2026-01-01',
}

Branching¶

from airflow.operators.python import BranchPythonOperator

def branch_func(**context):
    if context['execution_date'].day == 1:
        return 'first_day_task'
    return 'regular_task'

branch = BranchPythonOperator(
    task_id='branch',
    python_callable=branch_func,
)

first_day = PythonOperator(task_id='first_day_task', python_callable=...)
regular = PythonOperator(task_id='regular_task', python_callable=...)

branch >> [first_day, regular]

Dynamic Tasks¶

# Create tasks dynamically
task_names = ['extract', 'transform', 'load']

for name in task_names:
    PythonOperator(
        task_id=name,
        python_callable=lambda: print(f"Running {name}"),
    )

Practical Examples¶

Example 1 — ETL pipeline¶

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def extract():
    import pandas as pd
    df = pd.read_csv('/data/raw.csv')
    df.to_parquet('/data/staging/extracted.parquet')
    return len(df)

def transform():
    import pandas as pd
    df = pd.read_parquet('/data/staging/extracted.parquet')
    df = df.dropna()
    df.to_parquet('/data/staging/transformed.parquet')
    return len(df)

def load():
    import pandas as pd
    df = pd.read_parquet('/data/staging/transformed.parquet')
    # Load to database or data lake
    print(f"Loaded {len(df)} rows")

with DAG('etl_pipeline', start_date=datetime(2026, 1, 1), schedule_interval='@daily') as dag:
    t1 = PythonOperator(task_id='extract', python_callable=extract)
    t2 = PythonOperator(task_id='transform', python_callable=transform)
    t3 = PythonOperator(task_id='load', python_callable=load)

    t1 >> t2 >> t3

Example 2 — Multi-source extraction¶

def extract_from_source(source_name):
    def extract():
        print(f"Extracting from {source_name}")
        # Source-specific extraction logic
    return extract

sources = ['api', 'database', 'file']

with DAG('multi_source', start_date=datetime(2026, 1, 1)) as dag:
    extract_tasks = [
        PythonOperator(
            task_id=f'extract_{source}',
            python_callable=extract_from_source(source),
        )
        for source in sources
    ]

    # All sources feed into one transform
    transform = PythonOperator(task_id='transform', python_callable=transform)
    load = PythonOperator(task_id='load', python_callable=load)

    extract_tasks >> transform >> load

Example 3 — TaskFlow API (Airflow 2.x)¶

from airflow.decorators import dag, task

@dag(schedule_interval='@daily', start_date=datetime(2026, 1, 1))
def my_pipeline():

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

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

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

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

my_pipeline()

Common Patterns¶

Pattern: retry with backoff¶

default_args = {
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(minutes=60),
}

Pattern: alerting¶

from airflow.utils.email import send_email

def alert_failure(context):
    send_email(
        to=['team@example.com'],
        subject=f"Task failed: {context['task'].task_id}",
        html_content=f"DAG {context['dag'].dag_id} failed"
    )

default_args = {
    'on_failure_callback': alert_failure,
}

Pattern: data quality checks¶

def check_data_quality(**context):
    import pandas as pd
    df = pd.read_parquet('/data/output.parquet')
    assert len(df) > 0, "No data found"
    assert df.isnull().sum().sum() == 0, "Nulls found"
    print("Data quality check passed")

check = PythonOperator(
    task_id='quality_check',
    python_callable=check_data_quality,
)

Best Practices¶

Keep DAGs simple — one DAG per pipeline, not one massive DAG.
Use TaskFlow API — cleaner syntax for Airflow 2.x.
Set retries — transient failures are common.
Idempotent tasks — same inputs = same outputs, safe to rerun.
Use connections — never hardcode credentials.
Set timeouts — prevent hanging tasks.
Monitor task duration — alert on unexpected slowness.
Use variables carefully — they're global; prefer connections and params.

Common Mistakes¶

Mistake	Fix
Importing heavy libraries at DAG level	Import inside functions
Using `time.sleep()` in tasks	Use sensors instead
Hardcoding credentials	Use Airflow connections
No task timeout	Set `execution_timeout`
DAG doing too much	Split into smaller DAGs
Not using `depends_on_past` carefully	Avoid unless needed for ordering

Troubleshooting¶

Symptom	Cause / Fix
DAG not appearing	Check syntax; check `dags/` folder; restart scheduler
Task stuck in queued	Check worker availability; check resources
Task fails immediately	Check logs; check dependencies
Slow DAG	Check task durations; parallelize
Import errors	Check Python path; check installed packages

etl-elt.md — ETL/ELT patterns
data-pipelines.md — pipeline design patterns
delta-lake.md — Delta Lake operations
docker — containerizing Airflow
ci-cd — CI/CD for pipelines

Official Documentation¶

Apache Airflow — https://airflow.apache.org/docs/
Airflow TaskFlow API — https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html

Last Updated¶

2026-09-06 — Initial version.

Apache Airflow¶

Table of Contents¶

Installation / Setup¶

Core Syntax¶

DAG Definition¶

Operators¶

Task Dependencies¶

XCom (Cross-Communication)¶

Connections¶

Sensors¶

Templates and Jinja¶

Common Operations¶

Schedule Intervals¶

Context Variables¶

Branching¶

Dynamic Tasks¶

Practical Examples¶

Example 1 — ETL pipeline¶

Example 2 — Multi-source extraction¶

Example 3 — TaskFlow API (Airflow 2.x)¶

Common Patterns¶

Pattern: retry with backoff¶

Pattern: alerting¶

Pattern: data quality checks¶

Best Practices¶

Common Mistakes¶

Troubleshooting¶

Related Cheat Sheets¶

Official Documentation¶

Last Updated¶