Aufschiebbare Operatoren

Data-Pipelines mit Airflow aufbauen

Volker Janz

Senior Developer Advocate at Astronomer

Das Sensor-Problem

Sensoren belegen Worker-Slots

Poke-Modus: Sensor belegt den Worker-Slot durchgehend
Lang laufende Sensoren hungern andere Tasks aus

Reschedule-Modus

Poke-Modus (Standard)

from airflow.providers.standard.sensors.filesystem \
    import FileSensor

wait = FileSensor(
    task_id="wait",
    filepath="/data/report.csv",
    mode="poke",
    poke_interval=30,
)

Slot die ganze Zeit belegt

Reschedule-Modus

from airflow.providers.standard.sensors.filesystem \
    import FileSensor

wait = FileSensor(
    task_id="wait",
    filepath="/data/report.csv",
    mode="reschedule",
    poke_interval=300,
)

Slot zwischen Checks frei

Deferrable-Modus

Aufgeschobener Sensor wird vom Triggerer gehandhabt

Task übergibt an den Triggerer-Prozess
Triggerer nutzt async I/O für Hunderte paralleler Waits
Null Worker-Slots während des Wartens belegt

Deferrable aktivieren

from airflow.providers.standard.sensors.filesystem import FileSensor

wait_for_data = FileSensor(
    task_id="wait_for_data",
    filepath="/data/incoming/report.csv",

    deferrable=True,

    poke_interval=30,
    timeout=3600,
)

poke_interval und timeout gelten weiterhin

Welchen Modus wann nutzen

Modus	Am besten für	Worker-Slot
poke	Kurze Wartezeiten (< 1 Min.)	Durchgehend belegt
reschedule	Mittlere Wartezeiten, kein Triggerer	Zwischen Checks frei
deferrable	Lange Wartezeiten, viele Sensoren	Nie belegt

Läuft der Triggerer, ist deferrable fast immer die richtige Voreinstellung

Globale Deferrable-Einstellung

AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True

Alle kompatiblen Operatoren nutzen den Deferrable-Modus automatisch
Keine Änderungen am DAG-Code nötig
Pro Operator mit deferrable=False überschreibbar

Lass uns üben!

Data-Pipelines mit Airflow aufbauen