Apache Airflow Tutorial 2026: Production-DAGs richtig bauen

# Virtual Environment (Python 3.9–3.12)
python3 -m venv airflow-venv
source airflow-venv/bin/activate

# Airflow-Home-Verzeichnis setzen
export AIRFLOW_HOME=~/airflow

# Airflow 2.9+ mit Constraints für Reproduzierbarkeit
AIRFLOW_VERSION=2.9.3
PYTHON_VERSION=$(python3 -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')")
CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

# Metadata-DB initialisieren (SQLite für Dev, PostgreSQL für Prod)
airflow db migrate

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

# Webserver starten (Default: Port 8080)
airflow webserver --port 8080 -D

# Scheduler starten (eigenes Terminal oder Background)
airflow scheduler -D

# dags/weather_etl.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.http.sensors.http import HttpSensor
import requests
import json

# Default-Argumente für alle Tasks im DAG
default_args = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
}

def _extract_weather(**context):
    """Wetterdaten von Open-Meteo API extrahieren (kein API-Key nötig)."""
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": 52.52,
        "longitude": 13.41,
        "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum",
        "timezone": "Europe/Berlin",
        "past_days": 7,
    }
    response = requests.get(url, params=params, timeout=30)
    response.raise_for_status()
    data = response.json()
    # Push auf XCom für Downstream-Tasks
    context["ti"].xcom_push(key="weather_raw", value=data)

def _transform_weather(**context):
    """Rohe API-Response in Insert-Ready-Records umwandeln."""
    raw = context["ti"].xcom_pull(task_ids="extract_weather", key="weather_raw")
    daily = raw["daily"]
    records = []
    for i, date in enumerate(daily["time"]):
        records.append({
            "observation_date": date,
            "temp_max_c": daily["temperature_2m_max"][i],
            "temp_min_c": daily["temperature_2m_min"][i],
            "precipitation_mm": daily["precipitation_sum"][i],
            "city": "Berlin",
            "loaded_at": datetime.utcnow().isoformat(),
        })
    context["ti"].xcom_push(key="weather_records", value=records)

def _load_weather(**context):
    """Transformierte Records in PostgreSQL inserten."""
    from airflow.providers.postgres.hooks.postgres import PostgresHook
    records = context["ti"].xcom_pull(
        task_ids="transform_weather", key="weather_records"
    )
    hook = PostgresHook(postgres_conn_id="postgres_warehouse")
    # SQL-Dialekt: PostgreSQL
    insert_sql = """
        INSERT INTO weather_daily
            (observation_date, temp_max_c, temp_min_c, precipitation_mm, city, loaded_at)
        VALUES (%s, %s, %s, %s, %s, %s)
        ON CONFLICT (observation_date, city) DO UPDATE SET
            temp_max_c = EXCLUDED.temp_max_c,
            temp_min_c = EXCLUDED.temp_min_c,
            precipitation_mm = EXCLUDED.precipitation_mm,
            loaded_at = EXCLUDED.loaded_at;
    """
    for record in records:
        hook.run(insert_sql, parameters=(
            record["observation_date"],
            record["temp_max_c"],
            record["temp_min_c"],
            record["precipitation_mm"],
            record["city"],
            record["loaded_at"],
        ))

with DAG(
    dag_id="weather_etl_pipeline",
    default_args=default_args,
    description="Daily weather ETL von Open-Meteo API nach PostgreSQL",
    schedule="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["etl", "weather", "tutorial"],
) as dag:

    # Sensor: prüft, ob API erreichbar ist, bevor wir starten
    check_api = HttpSensor(
        task_id="check_api_available",
        http_conn_id="open_meteo_api",
        endpoint="v1/forecast?latitude=52.52&longitude=13.41&daily=temperature_2m_max",
        response_check=lambda response: response.status_code == 200,
        poke_interval=60,
        timeout=300,
    )

    # Zieltabelle anlegen wenn nicht da (SQL-Dialekt: PostgreSQL)
    create_table = PostgresOperator(
        task_id="create_table",
        postgres_conn_id="postgres_warehouse",
        sql="""
            CREATE TABLE IF NOT EXISTS weather_daily (
                observation_date DATE NOT NULL,
                temp_max_c FLOAT,
                temp_min_c FLOAT,
                precipitation_mm FLOAT,
                city VARCHAR(100) NOT NULL,
                loaded_at TIMESTAMP,
                PRIMARY KEY (observation_date, city)
            );
        """,
    )

    extract = PythonOperator(task_id="extract_weather", python_callable=_extract_weather)
    transform = PythonOperator(task_id="transform_weather", python_callable=_transform_weather)
    load = PythonOperator(task_id="load_weather", python_callable=_load_weather)

    # Task-Dependencies
    check_api >> create_table >> extract >> transform >> load

# Gängige Scheduling-Ausdrücke
schedule="@daily"          # Mitternacht UTC
schedule="@hourly"         # Volle Stunde
schedule="0 6 * * 1-5"    # 6 Uhr UTC, Werktage (Cron)
schedule="@weekly"         # Sonntag Mitternacht
schedule=None              # Nur extern getriggert (API, CLI, anderer DAG)

from airflow.timetables.datasets import DatasetOrTimeSchedule
from airflow.datasets import Dataset

# Täglich ODER wenn ein Upstream-Dataset aktualisiert wird
dag = DAG(
    dag_id="hybrid_schedule",
    schedule=DatasetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 6 * * *", timezone="UTC"),
        datasets=[Dataset("s3://bucket/raw/events/")],
    ),
)

# dags/weather_etl_taskflow.py
from datetime import datetime, timedelta
from airflow.decorators import dag, task
import requests

@dag(
    dag_id="weather_etl_taskflow",
    schedule="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
    },
    tags=["etl", "weather", "taskflow"],
)
def weather_etl_taskflow():

    @task()
    def extract() -> dict:
        """Wetterdaten holen. Return-Value wird automatisch auf XCom gepushed."""
        response = requests.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": 52.52,
                "longitude": 13.41,
                "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum",
                "timezone": "Europe/Berlin",
                "past_days": 7,
            },
            timeout=30,
        )
        response.raise_for_status()
        return response.json()

    @task()
    def transform(raw: dict) -> list[dict]:
        """Rohdaten in Records transformieren."""
        daily = raw["daily"]
        return [
            {
                "observation_date": daily["time"][i],
                "temp_max_c": daily["temperature_2m_max"][i],
                "temp_min_c": daily["temperature_2m_min"][i],
                "precipitation_mm": daily["precipitation_sum"][i],
                "city": "Berlin",
            }
            for i in range(len(daily["time"]))
        ]

    @task()
    def load(records: list[dict]):
        """Records in PostgreSQL laden via Hook."""
        from airflow.providers.postgres.hooks.postgres import PostgresHook
        hook = PostgresHook(postgres_conn_id="postgres_warehouse")
        for record in records:
            hook.run(
                """INSERT INTO weather_daily
                       (observation_date, temp_max_c, temp_min_c, precipitation_mm, city)
                   VALUES (%s, %s, %s, %s, %s)
                   ON CONFLICT (observation_date, city) DO UPDATE SET
                       temp_max_c = EXCLUDED.temp_max_c,
                       temp_min_c = EXCLUDED.temp_min_c,
                       precipitation_mm = EXCLUDED.precipitation_mm;""",
                parameters=(
                    record["observation_date"],
                    record["temp_max_c"],
                    record["temp_min_c"],
                    record["precipitation_mm"],
                    record["city"],
                ),
            )

    # Impliziter XCom: Return-Values fließen durch Funktions-Argumente
    raw_data = extract()
    transformed = transform(raw_data)
    load(transformed)

weather_etl_taskflow()

# CLI-Ansatz
airflow connections add 'postgres_warehouse' \
    --conn-type postgres \
    --conn-host warehouse.example.com \
    --conn-port 5432 \
    --conn-schema analytics \
    --conn-login etl_user \
    --conn-password 'securepassword'

# ENV-Ansatz (überschreibt DB-stored Connections)
export AIRFLOW_CONN_POSTGRES_WAREHOUSE='postgresql://etl_user:securepassword@warehouse.example.com:5432/analytics'

from airflow.models import Variable

# Schlecht: queried die DB bei jedem Scheduler-Parse (alle 30s by default)
threshold = Variable.get("quality_threshold")

# Gut: Zugriff innerhalb eines Tasks, läuft einmal pro Execution
@task()
def check_quality():
    threshold = float(Variable.get("quality_threshold", default_var="0.95"))
    # ... Validation-Logik

# SCHLECHT: läuft alle 30 Sekunden während des Parsings
import pandas as pd
data = pd.read_csv("s3://bucket/large_file.csv")  # Killt deinen Scheduler

# GUT: schwere Arbeit gehört in Tasks
@task()
def process_data():
    import pandas as pd
    data = pd.read_csv("s3://bucket/large_file.csv")

with DAG(
    dag_id="safe_dag",
    max_active_runs=3,         # Max 3 gleichzeitige Runs
    max_active_tasks=10,       # Max 10 Tasks über alle Runs
    # ...
) as dag:
    pass