Toán tử deferrable

Xây dựng Data Pipeline với Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Bài toán sensor

Cảm biến chiếm chỗ worker

Poke mode: sensor giữ chỗ worker suốt thời gian chờ
Sensor chạy lâu khiến tác vụ khác thiếu tài nguyên

Reschedule mode

Poke mode (mặc định)

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

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

Chỗ bị chiếm suốt thời gian chờ

Reschedule mode

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

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

Giải phóng chỗ giữa các lần kiểm tra

Deferrable mode

Sensor trì hoãn do Triggerer xử lý

Tác vụ bàn giao cho tiến trình Triggerer
Triggerer dùng async I/O để chờ đồng thời hàng trăm tác vụ
Không dùng chỗ worker trong lúc chờ

Bật chế độ 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 và timeout vẫn áp dụng

Khi nào dùng từng chế độ

Chế độ	Phù hợp nhất	Chỗ worker
poke	Chờ ngắn (< 1 phút)	Giữ suốt thời gian
reschedule	Chờ vừa, không có Triggerer	Giải phóng giữa các lần kiểm tra
deferrable	Chờ dài, nhiều sensor	Không bao giờ dùng

Nếu Triggerer đang chạy, deferrable gần như luôn là mặc định đúng

Thiết lập deferrable toàn cục

AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True

Tất cả operator tương thích sẽ tự động dùng chế độ deferrable
Không cần đổi mã Dag
Ghi đè theo operator với deferrable=False nếu cần

Cùng luyện tập nào!

Xây dựng Data Pipeline với Airflow