Apache Airflow: DAGs, Operators, and Scheduling

Apache Airflow: Building Production-Grade DAGs

Apache Airflow is the industry-standard open-source workflow orchestration platform for programmatically authoring, scheduling, and monitoring data pipelines. Originally developed at Airbnb in 2014, Airflow has become the backbone of data infrastructure at thousands of organizations.

Why Airflow Dominates Data Orchestration

Core Innovation:

• Treats workflows as code — Python-defined DAGs
• Encodes dependencies, scheduling, retry logic, and alerting
• Python-native authoring model leverages the full Python ecosystem

Key Capabilities:

1. Task dependencies — branching and joining workflows
2. Cross-task data passing — via XComs
3. Scheduling — time-based and event-driven
4. Distributed execution — across a worker pool
5. Production-grade UI — monitoring and manual intervention

Key Insight: Unlike cron-based scheduling or custom orchestration scripts, Airflow provides a rich set of primitives for expressing complex workflow logic.

Architecture Diagram

Key Concepts

Production DAG Example

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.email import EmailOperator
from airflow.sensors.s3_key import S3KeySensor
from airflow.utils.task_group import TaskGroup
from airflow.models import Variable
import logging

logger = logging.getLogger(__name__)

default_args = {
    "owner": "data-engineering",           # Team responsible for this DAG
    "retries": 3,                          # Auto-retry failed tasks up to 3 times
    "retry_delay": timedelta(minutes=5),   # Wait 5 minutes between retries
    "execution_timeout": timedelta(hours=2),  # Kill tasks after 2 hours
    "email_on_failure": True,              # Send email on final failure
    "email": ["data-alerts@company.com"],  # Alert recipients
    "on_failure_callback": lambda context: logger.error(
        f"Task {context['task_instance'].task_id} failed"
    ),
}


def extract_orders(**context):
    """
    Extract orders from the source database for the execution date.
    
    Parameters (via Airflow context):
        execution_date (datetime): The logical date for this DAG run
        ti (TaskInstance): Task instance for XCom operations
        
    Returns:
        list: Order records as list of dicts for XCom serialization
    """
    execution_date = context["execution_date"]
    import sqlalchemy

    engine = sqlalchemy.create_engine(
        Variable.get("source_db_connection")
    )
    query = f"""
        SELECT order_id, customer_id, amount, status, created_at
        FROM orders
        WHERE DATE(created_at) = '{execution_date.strftime('%Y-%m-%d')}'
    """
    df = pd.read_sql(query, engine)
    logger.info(f"Extracted {len(df)} orders for {execution_date}")

    # Store row count in XCom for downstream monitoring
    context["ti"].xcom_push(key="row_count", value=len(df))
    return df.to_dict("records")


def transform_orders(orders: list, **context):
    """
    Cleanse, deduplicate, and enrich order data.
    
    Parameters:
        orders (list): Raw order records from extract task
        context (dict): Airflow task context
        
    Returns:
        list: Transformed order records with customer tier enrichment
    """
    import pandas as pd

    df = pd.DataFrame(orders)
    df = df.drop_duplicates(subset=["order_id"])
    df = df[df["amount"] > 0]
    df["processed_at"] = datetime.utcnow()

    # Enrich with customer tier via merge
    customer_df = pd.read_parquet(
        Variable.get("customer_data_path")
    )
    df = df.merge(customer_df[["customer_id", "tier"]], on="customer_id", how="left")

    context["ti"].xcom_push(key="transformed_count", value=len(df))
    return df.to_dict("records")


def load_to_warehouse(records: list, **context):
    """
    Load transformed records into the data warehouse.
    
    Parameters:
        records (list): Transformed order records from transform task
        context (dict): Airflow task context
        
    Note: Uses chunked inserts (chunksize=1000) for memory efficiency
          with large datasets. For production, consider MERGE/UPSERT.
    """
    import sqlalchemy

    engine = sqlalchemy.create_engine(
        Variable.get("warehouse_connection")
    )
    df = pd.DataFrame(records)
    df.to_sql(
        "orders_fact",
        engine,
        if_exists="append",
        index=False,
        method="multi",     # Multi-row inserts for better performance
        chunksize=1000,     # Batch size to prevent memory issues
    )
    logger.info(f"Loaded {len(df)} records to warehouse")


with DAG(
    dag_id="daily_order_etl",
    default_args=default_args,
    description="Daily order extraction, transformation, and warehouse load",
    schedule_interval="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=["orders", "etl", "production"],
) as dag:

    # Sensor: wait for source data to be available in S3
    wait_for_data = S3KeySensor(
        task_id="wait_for_source_data",
        bucket_name="raw-data-lake",
        bucket_key="orders/{{ ds_nodash }}/orders.csv",
        timeout=3600,
        poke_interval=60,
        mode="reschedule",
    )

    with TaskGroup("etl_tasks") as etl_group:
        extract = PythonOperator(
            task_id="extract_orders",
            python_callable=extract_orders,
        )

        transform = PythonOperator(
            task_id="transform_orders",
            python_callable=transform_orders,
            op_kwargs={"orders": "{{ ti.xcom_pull(task_ids='etl_tasks.extract_orders') }}"},
        )

        load = PythonOperator(
            task_id="load_to_warehouse",
            python_callable=load_to_warehouse,
            op_kwargs={"records": "{{ ti.xcom_pull(task_ids='etl_tasks.transform_orders') }}"},
        )

        extract >> transform >> load

    notify_success = EmailOperator(
        task_id="notify_success",
        to=["data-alerts@company.com"],
        subject="Daily ETL completed for {{ ds }}",
        html_content="<p>Pipeline completed successfully.</p>",
        trigger_rule="all_success",
    )

    wait_for_data >> etl_group >> notify_success

Operator Comparison Table

Best Practices

1. Keep DAGs simple: Each DAG should represent a single logical workflow. Avoid nested SubDAGs; prefer TaskGroup for visual organization.
2. Use catchup=False in production to prevent unintended backfilling. Enable catchup only when explicitly backfilling historical data.
3. Set max_active_runs=1 for DAGs that modify shared state (e.g., database writes) to prevent race conditions.
4. Parameterize with Variables and Connections, not hardcoded values. Store secrets in the Airflow metadata database or a vault.
5. Implement idempotent tasks: every task must produce identical output when re-run with the same execution_date.
6. Use Sensors in reschedule mode to free up worker slots while waiting. poke mode holds a slot for the entire wait duration.
7. Configure execution_timeout on every task to prevent hung tasks from consuming resources indefinitely.
8. Monitor with callbacks: set on_failure_callback and on_retry_callback to integrate with Slack, PagerDuty, or email.
9. Test DAGs locally with airflow dags test before deploying to production. Validate task dependencies and data flow.
10. Version control DAGs using Git. Deploy via CI/CD pipelines that run linting, type checking, and unit tests.

Airflow Configuration Reference

See Also

• 018 - Advanced Airflow: XComs, Dynamic DAGs, and the TaskFlow API - Advanced patterns for complex workflows
• 029 - Prefect: Modern Workflow Orchestration - Alternative Python-native orchestration
• 016 - ETL vs ELT: Choosing the Right Approach - Pipeline paradigm selection
• 028 - Error Handling, Retries, and Dead Letter Queues - Production error handling patterns
• 027 - Pipeline Monitoring and Observability - Monitoring Airflow pipelines