Откладываемые операторы

Построение пайплайнов данных с Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Проблема сенсоров

Сенсоры, занимающие слоты воркеров

Режим poke: сенсор удерживает слот воркера всё время
Долгие сенсоры вытесняют другие задачи

Режим reschedule

Режим poke (по умолчанию)

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

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

Слот занят всё время

Режим reschedule

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

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

Слот освобождается между проверками

Откладываемый режим

Отложенный сенсор, обрабатываемый Triggerer

Задача передаётся процессу Triggerer
Triggerer использует асинхронный ввод-вывод для сотен параллельных ожиданий
Во время ожидания слоты воркеров не используются

Включение откладываемого режима

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 и timeout по-прежнему применяются

Когда использовать каждый режим

Режим	Лучший выбор	Слот воркера
poke	Короткие ожидания (< 1 мин)	Занят всё время
reschedule	Умеренные ожидания, без Triggerer	Освобождается между проверками
deferrable	Долгие ожидания, много сенсоров	Не используется

Если Triggerer запущен, deferrable — почти всегда правильный выбор по умолчанию

Глобальная настройка откладываемого режима

AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True

Все совместимые операторы автоматически используют откладываемый режим
Изменения в коде DAG не требуются
При необходимости задайте deferrable=False для конкретного оператора

Давайте потренируемся!

Построение пайплайнов данных с Airflow