Airflow DAG Orchestrator

Was macht Airflow DAG Orchestrator?

Entwerfen und betreiben Sie Apache Airflow DAGs für zuverlässige Daten-Pipeline-Orchestrierung mit ordnungsgemäßem Abhängigkeitsmanagement, SLAs und Überwachung.

Aktivierungsauslöser

Aktivieren bei: „Airflow“, „DAG“, „operator“, „sensor“, „scheduler“, „task dependency“, „SLA“, „backfill“, „XCom“, „TaskFlow API“, „MWAA“, „Cloud Composer“

NICHT für: dbt-Modellausführung → dbt-analytics-engineer (Airflow kann dbt zwar auslösen) | Stream-Verarbeitung → streaming-pipeline-architect | Workflow-Engine (Temporal) → distributed-transaction-manager

Schnellstart

1. DAG definieren – TaskFlow API verwenden (@dag, @task-Dekorateure) für Python-native DAGs
2. Zeitplan festlegen – Cron oder Timetable, mit catchup=False, sofern backfill nicht beabsichtigt ist
3. Wiederholungen konfigurieren – retries=2, retry_delay=timedelta(minutes=5) für jede Aufgabe
4. SLAs hinzufügen – sla=timedelta(hours=2) bei Aufgaben auf dem kritischen Pfad
5. Lokal testen – airflow dags test my_dag 2026-01-01 vor dem Bereitstellen

Kernfunktionen

Bereich	Technologien
Airflow	Apache Airflow 2.10+, MWAA, Cloud Composer 3
Operatoren	BashOperator, PythonOperator, KubernetesPodOperator
Provider	apache-airflow-providers-{snowflake, google, aws, dbt-cloud}
Executors	CeleryExecutor, KubernetesExecutor, LocalExecutor
Überwachung	SLA-Verstöße, Aufgabenlaufzeit, Airflow-Metriken → Prometheus

Architekturmuster

TaskFlow-API-DAG

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

@dag(
    schedule="0 6 * * *",          # täglich um 6 Uhr UTC
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "sla": timedelta(hours=2),
    },
    tags=["finance", "daily"],
)
def daily_revenue_pipeline():

    @task()
    def extract_payments() -> dict:
        """Extrahieren aus der Stripe-API"""
        data = stripe_client.list_payments(date=today())
        return {"count": len(data), "path": "s3://raw/payments/"}

    @task()
    def extract_orders() -> dict:
        """Extrahieren aus der Shopify-API"""
        data = shopify_client.list_orders(date=today())
        return {"count": len(data), "path": "s3://raw/orders/"}

    @task()
    def transform(payments: dict, orders: dict) -> str:
        """Verknüpfen und transformieren in DuckDB"""
        result_path = run_duckdb_transform(payments["path"], orders["path"])
        return result_path

    @task()
    def load(path: str):
        """Laden in Snowflake"""
        snowflake_copy_into("fct_revenue", path)

    # Abhängigkeiten über Funktionsaufrufe definieren
    payments = extract_payments()
    orders = extract_orders()
    transformed = transform(payments, orders)
    load(transformed)

daily_revenue_pipeline()

Dynamisches Task-Mapping (Fan-Out/Fan-In)

@task()
def get_partitions() -> list[str]:
    return ["2026-01-01", "2026-01-02", "2026-01-03"]

@task()
def process_partition(partition_date: str) -> dict:
    """Wird parallel für jede Partition ausgeführt"""
    return {"date": partition_date, "rows": process(partition_date)}

@task()
def aggregate(results: list[dict]):
    """Fan-In: erhält alle Partitionsergebnisse"""
    total = sum(r["rows"] for r in results)
    log.info(f"Verarbeitete insgesamt {total} Zeilen")

# Prozess_partition wird dynamisch über alle Partitionen gemappt
partitions = get_partitions()
results = process_partition.expand(partition_date=partitions)
aggregate(results)

dbt + Airflow-Integration

from airflow.operators.bash import BashOperator
from cosmos import DbtDag, ProjectConfig, ProfileConfig

# Option 1: cosmos (empfohlen)
dbt_dag = DbtDag(
    project_config=ProjectConfig("/opt/airflow/dbt/"),
    profile_config=ProfileConfig(
        profile_name="default",
        target_name="prod",
    ),
    schedule="@daily",
    dag_id="dbt_daily",
)

# Option 2: BashOperator (einfach)
dbt_run = BashOperator(
    task_id="dbt_run",
    bash_command="cd /opt/airflow/dbt && dbt build --select tag:daily",
)

Anti-Patterns

1. Schwere Aufgaben – Aufgaben sollten atomar sein; extrahieren+transformieren+laden nicht in eine Aufgabe packen
2. XCom für große Daten – XCom ist für Metadaten (Pfade, Zählungen); Daten über S3/GCS übergeben, nicht über XCom-Pickles
3. catchup=True aus Versehen – sofern Sie kein Backfill wünschen, setzen Sie catchup=False; andernfalls führt Airflow jedes verpasste Intervall aus
4. Keine Wiederholungen – temporäre Fehler sind häufig; immer retries >= 1 mit einer Verzögerung setzen
5. Top-Level-Code in DAG-Dateien – DAG-Dateien werden alle 30s analysiert; schwere Imports oder API-Aufrufe auf Modulebene verlangsamen den Scheduler

Qualitäts-Checkliste

• TaskFlow-API für Python-native DAGs verwendet (keine veraltete Operator-Verkettung)
• catchup=False, sofern Backfill nicht beabsichtigt ist
• Wiederholungen für alle Aufgaben konfiguriert (mindestens 1 Wiederholung)
• SLA für Aufgaben auf dem kritischen Pfad gesetzt
• XCom übergibt Referenzen (S3-Pfade), keine Daten-Nutzdaten
• Dynamisches Task-Mapping für Fan-Out-Parallelisierung verwendet
• DAG lokal mit airflow dags test vor dem Bereitstellen getestet
• Aufgaben-Idempotenz: erneutes Ausführen einer Aufgabe liefert das gleiche Ergebnis
• Alarmierung für Aufgabenfehler und SLA-Verstöße konfiguriert
• DAG-Tags zur Filterung in der Airflow-Benutzeroberfläche angewendet