Deferrable operators

Tvorba datových pipeline s Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Problém sensorů

Sensory obsazující worker sloty

Poke mode: sensor drží worker slot po celou dobu
Dlouho běžící sensory hladoví ostatní úlohy

Reschedule mode

Poke mode (výchozí)

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

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

Slot obsazen po celou dobu

Reschedule mode

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

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

Slot uvolněn mezi kontrolami

Deferrable mode

Odložený sensor zpracovaný Triggererem

Úloha předá řízení procesu Triggerer
Triggerer využívá async I/O pro stovky souběžných čekání
Během čekání se nepoužije žádný worker slot

Aktivace deferrable

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 a timeout stále platí

Kdy použít který mode

Mode	Vhodný pro	Worker slot
poke	Krátká čekání (< 1 min)	Obsazen po celou dobu
reschedule	Střední čekání, bez Triggereru	Uvolněn mezi kontrolami
deferrable	Dlouhá čekání, mnoho sensorů	Nikdy nepoužit

Pokud Triggerer běží, deferrable je téměř vždy správná volba

Globální nastavení deferrable

AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True

Všechny kompatibilní operátory použijí deferrable mode automaticky
Kód DAGu není potřeba měnit
Jednotlivé operátory lze přepsat pomocí deferrable=False

Pojďme cvičit!

Tvorba datových pipeline s Airflow