Deferrable operator

Operator peka waktu, seperti sensor dan operator yang menunggu peristiwa eksternal, menempati satu slot worker selama penungguan berlangsung. Mode deferrable menghilangkan biaya tersebut dengan melepas slot worker dari task yang sedang menunggu, sedangkan komponen triggerer melanjutkan eksekusi ketika peristiwa yang ditunggu terjadi. Topik ini membahas masalah slot worker yang terkunci selama penungguan, mekanisme mode deferrable beserta komponen triggerer, contoh sensor waktu yang berjalan dalam mode deferrable, dan perbandingannya dengan mode poke serta reschedule pada sensor.

Slot worker yang terkunci saat menunggu#

Setiap task yang sedang berjalan menempati satu slot worker. Sensor dalam mode poke mengunci slot itu sampai kondisinya terpenuhi, sehingga banyak DAG yang menunggu serentak dapat menghabiskan seluruh slot untuk pekerjaan yang tidak benar-benar berjalan. Dokumentasi Airflow mengilustrasikannya dengan hitungan sederhana. Apabila 100 slot worker tersedia dan 100 DAG sedang menunggu sensor tanpa mengerjakan apa pun, tidak ada slot tersisa untuk pekerjaan lain padahal kluster sedang kosong.

Mode reschedule pada sensor mengurangi masalah ini dengan melepas slot di antara pemeriksaan, tetapi pemeriksaan tetap dieksekusi oleh worker dan kelanjutan task hanya dapat dipicu oleh waktu. Untuk penungguan yang lebih panjang atau bergantung pada peristiwa lain, Airflow menyediakan mode deferrable yang melepas slot worker sepenuhnya sampai peristiwa pemicunya terjadi.

Mekanisme deferrable dan komponen triggerer#

Operator yang mendukung mode deferrable menerima argumen deferrable=True. Ketika task tersebut harus menunggu, task berhenti berjalan pada worker dan beralih ke status deferred. Pada status ini, proses task dihentikan, slot worker dilepas, dan penungguan diserahkan kepada objek trigger, yaitu potongan kode asinkron yang mengevaluasi kondisi tanpa menempati worker.

Objek trigger dijalankan oleh triggerer, yaitu proses Airflow yang dirancang untuk menampung banyak penungguan asinkron sekaligus. Triggerer terus mengevaluasi trigger yang terdaftar, dan ketika peristiwa tercapai, misalnya batas waktu tertentu sudah lewat, trigger mengirimkan peristiwa itu ke task. Task kemudian dijadwalkan kembali pada worker untuk melanjutkan eksekusinya dari titik pemberhentiannya.

Menulis operator deferrable sendiri berarti menulis kelas trigger yang tepat, dan cara penulisannya dijelaskan pada dokumentasi Airflow. Topik ini berfokus pada penggunaan operator siap pakai yang sudah mendukung mode tersebut.

Sensor waktu dalam mode deferrable#

TimeSensor dari provider standard menunggu sampai waktu tertentu dalam sehari terlewati. Kelas ini mendukung deferrable=True, sehingga task segera beralih ke status deferred dan penungguan waktunya diserahkan kepada triggerer. TimeSensorAsync adalah nama lama kelas ini dan kini berstatus deprecated, yakni masih ada tetapi tidak disarankan dan akan dihapus pada versi berikutnya. Bentuk yang disarankan adalah TimeSensor dengan argumen deferrable=True.

DAG berikut menunda pemrosesan sampai pukul 13.00.00 terlewati, lalu menjalankan task pemrosesannya.

"""DAG yang menunggu batas waktu harian dengan sensor deferrable."""
from datetime import time
from airflow.providers.standard.sensors.time import TimeSensor
from airflow.sdk import dag, task


@dag(
    dag_id="tunggu_jam_proses",
    schedule=None,
    description="Menunggu lewat pukul 13.00 UTC sebelum memulai pemrosesan.",
)
def tunggu_jam_proses():
    tunggu = TimeSensor(
        task_id="tunggu_lewat_jam",
        target_time=time(13, 0, 0),
        deferrable=True,
    )

    @task
    def mulai_pemrosesan() -> None:
        """Berjalan setelah batas waktu terlewati."""
        print("batas waktu terlewati, pemrosesan dimulai")

    tunggu >> mulai_pemrosesan()


dag = tunggu_jam_proses()

Task tunggu_lewat_jam berjalan terlebih dahulu dan langsung menyerahkan penungguannya kepada triggerer, sedangkan mulai_pemrosesan dijadwalkan setelah sensor sukses. Nilai target_time bertipe datetime.time, sehingga pukul 13.00.00 ditulis sebagai time(13, 0, 0).

# uji bahwa berkas DAG dapat diparse tanpa keluaran apa pun
python3 dag_tunggu_jam.py

# periksa kelas task dan status mode deferrable-nya
python3 -c "import dag_tunggu_jam; d = dag_tunggu_jam.dag; t = d.get_task('tunggu_lewat_jam'); print(type(t).__name__); print('deferrable =', t.deferrable)"

TimeSensor
deferrable = True

Keluaran menegaskan bahwa task sensor tersebut adalah TimeSensor biasa dengan atribut deferrable bernilai True, sehingga perilaku asinkronnya berasal dari argumen deferrable=True, bukan dari kelas yang berbeda. DAG ini tidak dijalankan dengan airflow dags test karena lingkungan latihan tidak menjalankan triggerer. Tanpa triggerer, task tunggu_lewat_jam beralih ke status deferred dan tidak pernah dilanjutkan, sehingga dag run tidak berakhir. Pengujian operator deferrable memerlukan deployment yang menjalankan triggerer, misalnya pemasangan lokal yang lengkap atau Docker Compose dengan layanan triggerer aktif.

Membandingkan poke, reschedule, dan deferrable#

Pemilihan mode mengikuti durasi penungguan dan sifat pemicunya. Mode poke tepat ketika kondisi berubah dalam hitungan detik dan penungguannya pendek, sehingga slot yang terkunci hanya terpakai sebentar. Mode reschedule tepat untuk pemeriksaan berbasis waktu dengan interval menit, sehingga slot dilepas di antara pemeriksaan. Mode deferrable tepat ketika penungguan bisa berlangsung lama atau banyak task menunggu bersamaan, dengan syarat triggerer tersedia. Dokumentasi Airflow menambahkan catatan khusus untuk sensor, yaitu mode reschedule dan deferrable sama-sama melepas slot worker, tetapi hanya mode deferrable yang bebas memilih peristiwa pemicunya melalui trigger.

Kesalahan umum#

Menganggap mode reschedule dan deferrable sebagai hal yang sama. Keduanya melepas slot worker, tetapi reschedule tetap menjalankan pemeriksaan pada worker secara berkala dan hanya mendukung pemicu berbasis waktu, sedangkan deferrable menunggu peristiwa pada triggerer. Penungguan berbasis waktu yang pendek biasanya cukup ditangani dengan mode reschedule tanpa trigger khusus.

Menjalankan DAG deferrable pada lingkungan tanpa triggerer. Task beralih ke status deferred dan tidak pernah dilanjutkan, sehingga dag run tampak menggantung tanpa pesan gagal yang jelas. Pastikan layanan triggerer berjalan sebelum menggunakan deferrable=True pada task mana pun.

Menggunakan nama kelas lama TimeSensorAsync pada contoh baru. Kelas tersebut berstatus deprecated dan akan dihapus pada versi provider berikutnya, sehingga DAG baru menggunakan TimeSensor dengan deferrable=True.

Ringkasan#

• Mode deferrable memindahkan task yang menunggu ke status deferred, sehingga slot worker dilepas sampai peristiwa pemicunya terjadi.
• Objek trigger yang asinkron dijalankan komponen triggerer, layanan terpisah yang wajib berjalan pada deployment yang menggunakan mode ini.
• Operator siap pakai mendukung mode ini melalui deferrable=True, dan TimeSensor menggantikan TimeSensorAsync yang berstatus deprecated.
• Mode reschedule juga melepas slot di antara pemeriksaan, tetapi kelanjutan task hanya dapat dipicu oleh waktu.

Referensi#

• Deferrable Operators & Triggers — Apache Airflow — Mekanisme deferral, peran triggerer, dan perbedaan mode reschedule dengan deferrable pada sensor.
• TimeSensor — Apache Airflow Standard Provider — Parameter TimeSensor dan TimeSensorAsync untuk menunggu waktu tertentu dalam sehari.
• Sensors — Apache Airflow — Mode poke dan reschedule beserta parameter bawaan BaseSensorOperator.