Claude Code & Apache Airflow: DAGs generieren, debuggen und skalieren

Apache Airflow ist das Rückgrat vieler Datenpipelines — aber wer schon einmal einen DAG mit zehn voneinander abhängigen Tasks, Retry-Logik, Sensors und variablen Schedules gebaut hat, weiß: die eigentliche Arbeit liegt im Detail. Wo genau soll der start_date liegen? Warum greift catchup nicht wie erwartet? Ist dieser Task wirklich idempotent?

Claude Code Airflow — also Claude Code als Entwicklungspartner für Airflow-Projekte — verändert diese Arbeit grundlegend. Nicht weil es Airflow ersetzt, sondern weil es den gesamten DAG-Code, die Konfiguration und die Fehlermeldungen gleichzeitig im Blick hat und konkrete, kontextbezogene Antworten liefert statt generischer Dokumentationsverweise.

1. Was ist Apache Airflow?

Apache Airflow ist eine Open-Source-Plattform zur Workflow-Orchestrierung. Du definierst Workflows als gerichtete azyklische Graphen — kurz DAGs (Directed Acyclic Graphs) — in Python. Airflow plant diese DAGs automatisch ein, führt sie aus und protokolliert Erfolg oder Fehler jedes einzelnen Tasks.

Die Kernkomponenten:

• DAG — der Workflow selbst: eine Python-Datei, die Tasks und ihre Abhängigkeiten beschreibt
• Scheduler — prüft kontinuierlich, welche DAG-Runs fällig sind, und stellt Tasks in die Warteschlange
• Executor — führt Tasks aus (lokal, via Celery-Worker oder in Kubernetes-Pods)
• Webserver — die UI: DAG-Übersicht, Logs, manuelle Trigger
• Metadaten-Datenbank — speichert DAG-Runs, Task-States und XCom-Werte

Airflow eignet sich besonders für ETL-Pipelines, ML-Training-Workflows, Report-Generierung und alle regelmäßigen Batch-Prozesse, bei denen Reihenfolge und Fehlerbehandlung wichtig sind.

2. DAG-Definition: Tasks, Operators und Dependencies

Ein einfacher DAG in der klassischen Airflow-Syntax:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "data-team",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
}

with DAG(
    dag_id="etl_orders",
    default_args=default_args,
    schedule_interval="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    extract = PythonOperator(
        task_id="extract_orders",
        python_callable=extract_orders_from_api,
    )

    transform = PythonOperator(
        task_id="transform_orders",
        python_callable=transform_to_schema,
    )

    load = BashOperator(
        task_id="load_to_warehouse",
        bash_command="python /opt/scripts/load_warehouse.py",
    )

    extract >> transform >> load

Der >>-Operator definiert Abhängigkeiten: transform startet erst, wenn extract erfolgreich war. Für parallele Tasks schreibt man [task_a, task_b] >> task_c — task_c wartet auf beide.

3. Scheduling: schedule_interval, start_date, catchup und Backfill

Das Scheduling ist der häufigste Stolperstein in Airflow. Die wichtigsten Parameter:

• schedule_interval — Cron-Expression oder Preset (@daily, @hourly). Airflow verwendet das Ende eines Intervalls als Trigger: Ein DAG mit @daily und start_date=2026-01-01 läuft erstmals am 2. Januar um 00:00 Uhr.
• start_date — Startpunkt des Schedulings. Immer in der Vergangenheit setzen, nie datetime.now() verwenden — das führt zu nicht-deterministischen DAG-Runs.
• catchup — ob Airflow rückwirkend alle verpassten Intervalle zwischen start_date und heute aufholt. Standard ist True; in den meisten Produktionsfällen will man catchup=False.
• Backfill — gezieltes Nachholen einzelner Zeiträume via CLI: airflow dags backfill -s 2026-08-01 -e 2026-08-31 etl_orders

4. TaskFlow API: @dag, @task, XComs und Sensors

Ab Airflow 2.0 gibt es die TaskFlow API, die DAGs deutlich lesbarer macht. Statt Operator-Instanzen zu erstellen, dekorierst du Python-Funktionen direkt:

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

@dag(
    schedule_interval="@daily",
    start_date=datetime(2026, 1, 1),
    catchup=False,
)
def etl_orders_taskflow():

    @task()
    def extract() -> dict:
        return {"orders": fetch_from_api()}

    @task()
    def transform(raw: dict) -> list:
        return [clean(order) for order in raw["orders"]]

    @task()
    def load(records: list) -> None:
        insert_to_warehouse(records)

    raw = extract()
    cleaned = transform(raw)
    load(cleaned)

etl_orders_taskflow()

Der Rückgabewert eines @task wird automatisch als XCom gespeichert und als Argument an den nächsten Task übergeben. XComs ermöglichen so den Datenaustausch zwischen Tasks — für große Datenmengen sollte man stattdessen Pfade oder Referenzen übergeben und die eigentlichen Daten in S3 oder ähnlichem ablegen.

Sensors sind spezialisierte Operators, die auf ein externes Ereignis warten: FileSensor (Datei vorhanden?), HttpSensor (Endpunkt antwortet?), ExternalTaskSensor (anderer DAG abgeschlossen?). Mit mode="reschedule" geben Sensors den Worker-Slot frei, während sie warten — wichtig bei langen Wartezeiten.

5. Connections, Variables, Hooks und Executor-Typen

Airflow bietet drei Mechanismen für Konfiguration und externe Zugriffe:

• Connections — gespeicherte Verbindungsdaten (Host, Port, Login, Password) für Datenbanken, APIs und Cloud-Dienste. Im Code abrufbar via BaseHook.get_connection("my_conn_id").
• Variables — Key-Value-Paare für DAG-übergreifende Konfiguration: Variable.get("s3_bucket_name"). Für sensible Werte Secrets-Backend (AWS Secrets Manager, HashiCorp Vault) nutzen.
• Hooks — wiederverwendbare Klassen, die eine Connection kapseln. PostgresHook gibt z. B. direkt eine Datenbankverbindung zurück, ohne dass man Credentials im DAG-Code schreibt.

Die Wahl des Executors bestimmt, wie Tasks ausgeführt werden:

• LocalExecutor — Tasks laufen als Subprozesse auf dem Scheduler-Host. Einfach einzurichten, für kleine bis mittlere Workloads.
• CeleryExecutor — Tasks werden an externe Worker-Prozesse verteilt (Redis oder RabbitMQ als Broker). Skalierbar horizontal, komplexer im Betrieb.
• KubernetesExecutor — jeder Task bekommt einen eigenen Pod im Kubernetes-Cluster. Maximale Isolation und Skalierbarkeit, höherer Overhead pro Task.

6. Claude Code Airflow: Konkrete Tipps

DAGs generieren lassen

Statt einen DAG von Grund auf zu schreiben, beschreibe den Workflow in Prosaform und lass Claude Code den DAG generieren:

claude "Erstelle einen Airflow-DAG mit TaskFlow API für folgenden Workflow:
1. Täglich um 05:30 UTC starten
2. Bestellungen der letzten 24h aus PostgreSQL (Connection: postgres_prod) laden
3. Auf Vollständigkeit prüfen (mindestens 100 Zeilen, sonst Task fehlschlagen)
4. In Parquet nach S3 (Variable: s3_reports_bucket) schreiben
5. Slack-Benachrichtigung senden
catchup=False, 3 Retries mit 10 Minuten Delay, idempotent."

Claude Code kennt die aktuellen Provider-Pakete, schreibt die Decorator-Syntax korrekt und baut Idempotenz direkt ein — z. B. indem es den S3-Pfad mit dem logical_date-Parameter parametrisiert, sodass wiederholte Runs denselben Pfad überschreiben statt neue Dateien anzulegen.

TaskFlow-Migration bestehender DAGs

Ältere Airflow-DAGs mit klassischen Operators lassen sich einfach migrieren:

claude "Migriere diesen DAG auf die TaskFlow API (@dag, @task).
Behalte alle Abhängigkeiten, Retry-Konfigurationen und XCom-Übergaben bei." < legacy_dag.py

Claude Code liest den bestehenden DAG, mappt jeden PythonOperator auf einen @task-Decorator, ersetzt explizite xcom_push/xcom_pull-Aufrufe durch Rückgabewerte und Parameter — und prüft dabei, ob die Abhängigkeiten korrekt abgebildet sind.

Scheduling-Fehler debuggen

Scheduling-Probleme sind in Airflow oft schwer zu diagnostizieren, weil die Fehlermeldungen im Scheduler-Log vergraben sind:

claude "Mein DAG 'etl_orders' wird nicht getriggert obwohl der Schedule
'0 6 * * *' konfiguriert ist. start_date ist datetime(2026, 9, 1).
Hier ist die DAG-Datei und der relevante Abschnitt aus dem Scheduler-Log." < dag_and_log.txt

Claude Code prüft typische Ursachen systematisch: start_date in der Zukunft, falsch konfiguriertes Timezone-Handling (Airflow arbeitet intern immer UTC), catchup=True mit einem weit zurückliegenden start_date der hunderte Runs erzeugt, oder ein DAG der auf paused=True steht.

Idempotenz sicherstellen

Idempotente Tasks sind das Fundament zuverlässiger Pipelines: ein Task, der mehrfach mit denselben Parametern ausgeführt wird, muss immer dasselbe Ergebnis produzieren — ohne Duplikate, ohne Datenverlust.

claude "Prüfe diesen Task auf Idempotenz-Probleme und erkläre,
wie ich ihn absichern kann:" < load_task.py

Claude Code erkennt klassische Idempotenz-Brüche: INSERT ohne ON CONFLICT DO UPDATE, Datei-Writes ohne vorheriges Delete, State der zwischen Runs akkumuliert. Die Lösung ist fast immer eine Kombination aus logical_date-parametrisierten Pfaden, Upsert-Logik in der Datenbank und explizitem Cleanup am Anfang jedes Tasks.

"Idempotenz ist keine Optimierung — sie ist die Grundbedingung dafür, dass Retry-Logik und Backfills überhaupt funktionieren. Ein Task, der beim zweiten Run Duplikate erzeugt, ist ein kaputtes Fundament."

Verwandte Artikel, die auf diesem Thema aufbauen:

• Claude Code für Python-Projekte — Python-Workflows, Tests und Refactoring mit Claude Code
• Claude Code Debugging — Stack Traces und Logs systematisch analysieren