Відкладні оператори

Створення конвеєрів даних з 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,
)

Слот звільняється між перевірками

Режим deferrable

Відкладений сенсор обробляє Triggerer

Завдання передається процесу Triggerer
Triggerer використовує асинхронний I/O для сотень паралельних очікувань
Під час очікування використано нуль робочих слотів

Увімкнення 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 і timeout усе ще діють

Коли який режим використовувати

Режим	Найкраще для	Робочий слот
poke	Короткі очікування (< 1 хв)	Зайнятий увесь час
reschedule	Помірні очікування, без Triggerer	Вільний між перевірками
deferrable	Довгі очікування, багато сенсорів	Не використовується

Якщо Triggerer запущено, deferrable майже завжди найкращий типовий вибір

Глобальне налаштування deferrable

AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True

Усі сумісні оператори автоматично працюють у режимі deferrable
Змін у коді Dag не потрібно
За потреби перевизначте для оператора через deferrable=False

Давайте потренуємось!

Створення конвеєрів даних з Airflow