airflow

Airflow

Apache Airflow is the industry standard for programmatic data pipeline orchestration, featuring dynamic DAG generation, event-driven triggers, and robust task scheduling.

When to Use

• Enterprise Data Pipeline Orchestration: Scheduling, monitoring, and authoring complex DAGs across cloud and on-premise systems.
• Modern TaskFlow API Workloads: Writing clean, Pythonic DAGs using @task and @dag decorators with automated XCom serialization.
• Dynamic Task Mapping: Fan-out and fan-in workflows processing variable numbers of files, partitions, or microservices.
• Event-Driven & Deferrable Operators: Minimizing worker slot utilization during long-running external jobs with async triggers.

Quick Start

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

@dag(
    schedule="@daily",
    start_date=datetime(2025, 1, 1),
    catchup=False,
    tags=["pipeline"]
)
def etl_pipeline():
    @task
    def extract() -> list[dict]:
        return [{"id": 1, "val": 100}, {"id": 2, "val": 200}]

    @task
    def transform(data: list[dict]) -> list[dict]:
        return [{**d, "val": d["val"] * 2} for d in data]

    @task
    def load(data: list[dict]):
        print(f"Loaded {len(data)} transformed records.")

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

etl_pipeline()

Core Concepts

TaskFlow API & Functional DAG Authoring

Pythonic DAG definition with automatic XCom data passing:

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

default_args = {
    'owner': 'data-platform',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

@dag(
    dag_id='customer_metrics_pipeline',
    default_args=default_args,
    start_date=datetime(2026, 1, 1),
    schedule='@daily',
    catchup=False,
    tags=['analytics', 'production']
)
def customer_metrics_dag():

    @task
    def extract_raw_records() -> list[dict]:
        return [
            {'user_id': 101, 'spend': 120.50},
            {'user_id': 102, 'spend': 450.00},
            {'user_id': 103, 'spend': 89.20},
        ]

    @task
    def compute_summary(records: list[dict]) -> dict:
        total = sum(r['spend'] for r in records)
        count = len(records)
        return {'total_spend': total, 'avg_spend': total / count, 'count': count}

    @task
    def publish_metrics(summary: dict):
        print(f"Published KPI: Total=${summary['total_spend']}, Avg=${summary['avg_spend']:.2f}")

    raw = extract_raw_records()
    summary = compute_summary(raw)
    publish_metrics(summary)

customer_pipeline = customer_metrics_dag()