Operadores aplazables

Creación de canalizaciones de datos con Airflow

Volker Janz

Senior Developer Advocate at Astronomer

El problema de los sensores

Sensores ocupando ranuras de worker

Modo poke: el sensor retiene la ranura del worker todo el tiempo
Sensores de larga duración asfixian otras tareas

Modo reschedule

Modo poke (predeterminado)

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

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

Ranura ocupada todo el tiempo

Modo reschedule

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

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

Ranura liberada entre comprobaciones

Modo aplazable

Sensor aplazado gestionado por Triggerer

La tarea se delega al proceso Triggerer
Triggerer usa E/S asíncrona para cientos de esperas concurrentes
Cero ranuras de worker usadas durante la espera

Activar modo aplazable

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 y timeout siguen aplicando

Cuándo usar cada modo

Mode	Best for	Worker slot
poke	Esperas cortas (< 1 min)	Ocupada todo el tiempo
reschedule	Esperas medias, sin Triggerer	Liberada entre comprobaciones
deferrable	Esperas largas, muchos sensores	Nunca usada

Si Triggerer está en ejecución, deferrable casi siempre es la opción por defecto correcta

Ajuste global de modo aplazable

AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True

Todos los operadores compatibles usan el modo aplazable automáticamente
No hace falta cambiar el código del DAG
Puedes anular por operador con deferrable=False si hace falta

¡Vamos a practicar!

Creación de canalizaciones de datos con Airflow