Deferrable operators

การสร้าง Data Pipeline ด้วย Airflow

Volker Janz

Senior Developer Advocate at Astronomer

ปัญหาของ sensor

Sensors ที่ครอบครอง worker slot

Poke mode: sensor ถือครอง worker slot ตลอดเวลา
Sensor ที่รันนานจะ แย่งทรัพยากร ของงานอื่น

Reschedule mode

Poke mode (ค่าเริ่มต้น)

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

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

ถือครอง slot ตลอดเวลา

Reschedule mode

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

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

คืน slot ระหว่างการตรวจสอบ

Deferrable mode

Deferred sensor ที่จัดการโดย Triggerer

งานส่งต่อให้กระบวนการ Triggerer
Triggerer ใช้ async I/O รองรับการรอพร้อมกันได้หลายร้อยรายการ
ระหว่างรอ ไม่ใช้ worker slot เลย

การเปิดใช้งาน 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 ยังคงมีผล

เลือกใช้ mode ไหนดี

Mode	เหมาะสำหรับ	Worker slot
poke	รอเวลาสั้น (< 1 นาที)	ถือครองตลอดเวลา
reschedule	รอปานกลาง ไม่มี Triggerer	คืนระหว่างตรวจสอบ
deferrable	รอนาน, sensor จำนวนมาก	ไม่ใช้เลย

หาก Triggerer ทำงานอยู่ deferrable มักเป็นตัวเลือกที่ดีที่สุดเสมอ

การตั้งค่า deferrable แบบ global

AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True

operator ทุกตัวที่รองรับจะใช้ deferrable mode โดยอัตโนมัติ
ไม่ต้องแก้ไขโค้ด DAG
หากต้องการ ให้ override รายตัวด้วย deferrable=False

มาฝึกกันเถอะ!

การสร้าง Data Pipeline ด้วย Airflow