Operadores adiáveis

Criando Pipelines de Dados com Airflow

Volker Janz

Senior Developer Advocate at Astronomer

O problema dos sensores

Sensores ocupando slots de worker

Modo poke: o sensor mantém o slot de worker o tempo todo
Sensores de longa duração prejudicam outras tarefas

Modo reschedule

Modo poke (padrão)

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

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

Slot ocupado o tempo todo

Modo reschedule

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

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

Slot liberado entre verificações

Modo adiável

Sensor adiado tratado pelo Triggerer

A tarefa repassa para o processo Triggerer
O Triggerer usa E/S assíncrona para centenas de esperas simultâneas
Zero slots de worker usados durante a espera

Ativando o modo adiável

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 e timeout ainda se aplicam

Quando usar cada modo

Modo	Melhor para	Slot de worker
poke	Esperas curtas (< 1 min)	Mantido o tempo todo
reschedule	Esperas moderadas, sem Triggerer	Liberado entre verificações
deferrable	Esperas longas, muitos sensores	Nunca usado

Se o Triggerer estiver em execução, adiável quase sempre é o padrão certo

Configuração global do modo adiável

AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True

Todos os operadores compatíveis usam o modo adiável automaticamente
Sem mudanças no código do Dag
Substitua por operador com deferrable=False se precisar

Vamos praticar!

Criando Pipelines de Dados com Airflow