Unknown-333/authoring-airflow-dags

Authoring Airflow DAGs

Wann verwenden

• Erstellen oder Refaktorieren von Airflow-DAGs und Tasks.
• Konfigurieren von Zeitplänen, Catchup/Backfill, Wiederholungen und SLAs.
• Daten zwischen Tasks übergeben (XCom) oder Verbindungen/Variablen nutzen.
• NICHT verwenden zur Diagnose eines fehlerhaften, laufenden DAGs (dafür debugging-airflow-pipelines nutzen).

Arbeitsablauf

- [ ] Jeden Task idempotent und durch das Datenintervall parametrisiert machen
- [ ] Teure/importlastige Code-Abschnitte innerhalb der Tasks belassen, nicht auf Modulebene
- [ ] Zeitplan + Catchup bewusst setzen
- [ ] Wiederholungen, retry_delay und SLAs konfigurieren
- [ ] Abhängigkeiten über TaskFlow-Rückgabewerte oder >>-Operatoren verknüpfen

1. Idempotente Tasks – Ein Task für das Intervall 2026-01-15 muss das gleiche Ergebnis liefern, unabhängig davon, ob er einmal oder erneut ausgeführt wird. Das Datenintervall verwenden, nicht datetime.now().
2. Kein schwerer Top-Level-Code – Der Scheduler parst jede DAG-Datei häufig; Datenbankaufrufe, API-Aufrufe oder große Imports auf Modulebene verlangsamen die Planung und können das Parsen unterbrechen. Diese innerhalb der Tasks platzieren.
3. Zeitplan + Catchup bewusst setzen – catchup=True führt für jedes verpasste Intervall seit start_date einen Backfill durch; standardmäßig auf False setzen, es sei denn, dies ist gewünscht.
4. Wiederholungen und SLAs – Vorübergehende Fehler sind normal; retries und retry_delay setzen; SLAs/Benachrichtigungen für Verspätungen nutzen.

Muster

TaskFlow DAG, idempotent und sauber verknüpft:

from airflow.decorators import dag, task
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5)},
    tags=["orders"],
)
def orders_pipeline():

    @task
    def extract(data_interval_start=None):
        # Das Intervall verwenden, nicht now(), damit Wiederholungen deterministisch sind.
        return fetch_orders(day=data_interval_start.date())

    @task
    def load(rows):
        # Partition löschen-einfügen -> idempotent bei Wiederholung.
        overwrite_partition("fct_orders", rows)

    load(extract())

orders_pipeline()

Kleine Daten via XCom übergeben (Rückgabewerte); große Daten via Storage übergeben – auf S3/GCS/Data Warehouse schreiben und den Pfad/Schlüssel übergeben, niemals Megabyte durch XCom.

Verbindungen/Variablen für Geheimnisse und Konfiguration nutzen (BaseHook.get_connection, Variable.get), niemals hartcodierte Anmeldedaten.

Häufige Fallstricke

• Top-Level-API-/DB-Aufrufe oder schwere Imports – verlangsamen den Scheduler und können das DAG-Parsen in der gesamten Bereitstellung zum Scheitern bringen.
• datetime.now() innerhalb von Tasks – bricht Idempotenz und Backfills; data_interval_start/_end verwenden.
• catchup=True unbeabsichtigt – überschwemmt den Cluster beim ersten Deployment mit historischen Läufen.
• Große Payloads durch XCom – bläht die Metadaten-Datenbank auf; Referenzen übergeben.
• Dynamisches start_date (z.B. days_ago) – macht Zeitpläne nichtdeterministisch; einen festen Zeitstempel verwenden.
• Ein Monstertask – Extract/Transform/Load aufteilen, damit Wiederholungen granulär sind.

Referenzen

• Referenz zu Zeitplanung, Catchup und Backfill