Sensor

Sensor adalah task yang menunggu suatu kondisi eksternal terpenuhi sebelum alur pipeline dilanjutkan, misalnya kedatangan berkas, kemunculan partisi data, atau selesainya pekerjaan pada sistem lain. Topik ini membahas cara kerja BaseSensorOperator, mode poke dan reschedule beserta dampak keduanya terhadap slot worker, parameter penanganan batas waktu, serta keluarga sensor yang tersedia dari berbagai provider. Hasil akhirnya adalah DAG berisi FileSensor yang menunggu berkas masuk. DAG tersebut dijalankan secara nyata dengan CLI, dan task hilirnya berhasil menghitung jumlah baris berkas itu.

Cara kerja sensor#

Semua sensor pada Airflow mewarisi BaseSensorOperator, kelas dasar yang mendefinisikan perilaku menunggu. Sensor mengeksekusi fungsi poke secara berulang. Fungsi ini menerima konteks task dan mengembalikan nilai boolean. Nilai True menyatakan kondisi telah terpenuhi sehingga sensor berstatus success dan task hilir boleh dijadwalkan, sedangkan nilai False membuat sensor menunggu poke_interval detik sebelum memeriksa lagi. Apabila batas timeout tercapai tanpa kondisi terpenuhi, sensor berstatus failed.

Kelas sensor siap pakai menyederhanakan fungsi poke menjadi parameter yang mudah digunakan, misalnya filepath pada FileSensor. Kondisi khusus yang tidak disediakan pustaka dapat ditulis melalui parameter python_callable pada BaseSensorOperator, yaitu fungsi Python biasa yang menjadi inti pemeriksaan. Pemeriksaan sederhana juga dapat ditulis sebagai @task biasa yang melakukan perulangan sampai kondisinya terpenuhi, apabila fitur tambahan sensor seperti soft_fail memang tidak dibutuhkan.

Beberapa parameter BaseSensorOperator mengatur biaya dan akibat dari penungguan. Parameter timeout membatasi total durasi penungguan sejak percobaan eksekusi pertama. Apabila batas itu tercapai dan soft_fail=True diaktifkan, sensor berstatus skipped, bukan failed, sehingga alur yang menoleransi ketiadaan data tidak ikut gagal. Parameter exponential_backoff menaikkan jeda antar pemeriksaan secara eksponensial hingga batas max_wait, sehingga sumber eksternal yang lambat tidak diperiksa terlalu sering.

Mode poke dan reschedule#

Mode penungguan menentukan komponen yang menahan sumber daya selama kondisi yang ditunggu belum terpenuhi. Mode poke, yaitu mode bawaan, menempati satu slot worker selama penungguan karena proses task tetap berjalan pada worker dan tidak mengerjakan apa pun di antara pemeriksaan. Mode reschedule melepas slot tersebut di antara pemeriksaan, lalu pemeriksaan berikutnya dijadwalkan ulang setelah poke_interval berlalu. Menurut dokumentasi Airflow, pemilihan mode pada dasarnya soal latensi, yaitu seberapa cepat kondisi terdeteksi. Pemeriksaan tiap detik lebih cocok pada mode poke, sedangkan pemeriksaan tiap menit lebih cocok pada mode reschedule.

Mode reschedule menghemat slot, tetapi pemeriksaan tetap dieksekusi oleh worker dan kelanjutan task hanya dapat dipicu oleh waktu. Untuk penungguan yang panjang atau bergantung pada peristiwa selain waktu, mode deferrable melepas slot worker sepenuhnya dengan menyerahkan penungguan kepada komponen triggerer. Mekanismenya dibahas pada topik tersendiri.

Menunggu berkas dengan FileSensor#

FileSensor dari provider standard memeriksa keberadaan berkas pada sistem berkas. Parameter filepath menyatakan berkas yang ditunggu dan mendukung template, sehingga jalurnya dapat mengikuti konteks eksekusi. Sensor ini membaca sistem berkas melalui koneksi bernama fs_default. Pada latihan topik ini, koneksi tersebut dideklarasikan melalui variabel lingkungan AIRFLOW_CONN_FS_DEFAULT agar tidak perlu dimasukkan ke basis data metadata.

DAG berikut menunggu berkas /tmp/ra-sensor/berkas-masuk.csv, kemudian menghitung jumlah barisnya. Sensor menggunakan poke_interval=1 dan timeout=10, sehingga apabila berkas tidak datang dalam sepuluh detik, sensor berstatus failed.

"""DAG yang menunggu kedatangan berkas masuk sebelum memprosesnya."""
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.sdk import dag, task


@dag(
    dag_id="tunggu_berkas",
    schedule=None,
    description="Menunggu berkas masuk lalu menghitung jumlah barisnya.",
)
def tunggu_berkas():
    tunggu = FileSensor(
        task_id="tunggu_berkas_masuk",
        filepath="/tmp/ra-sensor/berkas-masuk.csv",
        poke_interval=1,
        timeout=10,
        mode="poke",
    )

    @task
    def hitung_baris() -> None:
        """Menghitung jumlah baris berkas yang telah datang."""
        with open("/tmp/ra-sensor/berkas-masuk.csv") as berkas:
            print(f"berkas masuk berisi {sum(1 for _ in berkas)} baris")

    tunggu >> hitung_baris()


dag = tunggu_berkas()

Argumen mode="poke" tetap ditulis walaupun merupakan nilai bawaan agar pilihan mode penungguan terbaca langsung dari kode. Task hitung_baris berjalan hanya setelah sensor sukses karena dependensi tunggu >> hitung_baris() menyatakan urutan eksekusinya. Pola ini merupakan pola dasar pipeline peka data, yaitu sensor sebagai gerbang di hulu dan pekerjaan pengolahan di hilir.

# lingkungan latihan terpisah untuk topik ini
export AIRFLOW_HOME=/tmp/ra-sensor
export AIRFLOW__CORE__LOAD_EXAMPLES=False
# koneksi sistem berkas fs_default dibaca dari variabel lingkungan
export AIRFLOW_CONN_FS_DEFAULT='fs:///'

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

# periksa kelas task sensor dan parameternya tanpa menjalankan task
python3 -c "import dag_tunggu_berkas; d = dag_tunggu_berkas.dag; t = d.get_task('tunggu_berkas_masuk'); print(d.dag_id); print(type(t).__name__, t.poke_interval, t.timeout, t.mode)"

tunggu_berkas
FileSensor 1.0 10.0 poke

Keluaran menegaskan dua hal. Baris pertama menyatakan dag_id sesuai decorator, sedangkan baris kedua memperlihatkan bahwa task sensor adalah FileSensor dengan interval pemeriksaan 1 detik, batas waktu tunggu 10 detik, dan mode poke.

Menjalankan sensor secara lokal#

Sensor dijalankan dengan dags test sebagaimana DAG lain, dengan satu kebutuhan tambahan, yaitu kondisi yang ditunggu harus dipenuhi sebelum eksekusi. Berkas masuk dibuat terlebih dahulu di /tmp/ra-sensor, sehingga pemeriksaan pertama sensor langsung berhasil dan latihan tidak menunggu sama sekali.

# siapkan folder dags lalu salin berkas DAG ke dalamnya
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_tunggu_berkas.py "$AIRFLOW_HOME/dags/"

# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
airflow db migrate > /dev/null 2>&1

# berkas masuk dibuat lebih dulu agar pemeriksaan sensor langsung berhasil
mkdir -p /tmp/ra-sensor
printf "tanggal,suhu\n2026-11-04,24.5\n2026-11-05,25.0\n" > /tmp/ra-sensor/berkas-masuk.csv

# jalankan seluruh task DAG sekali secara lokal
airflow dags test tunggu_berkas 2026-11-04 > /tmp/ra-sensor/dags-test.log 2>&1

Keluaran perintah terakhir panjang dan bertanda waktu, sehingga tidak ditampilkan. Pada log, task tunggu_berkas_masuk mencatat pesan Success criteria met. Exiting. pada pemeriksaan pertamanya karena berkas sudah ada. Task tersebut langsung berstatus success tanpa satu kali pun menunggu. Task hitung_baris berjalan setelahnya dan mencetak berkas masuk berisi 3 baris, sedangkan dag run ditutup dengan status success. Apabila berkas masuk dihapus dan DAG dijalankan ulang tanpa berkas tersebut, sensor memeriksa tiap satu detik sampai batas sepuluh detik tercapai, lalu berstatus failed dengan error AirflowSensorTimeout.

Keluarga sensor lainnya#

ExternalTaskSensor dari provider standard menunggu selesainya task atau dag run pada DAG lain. Sensor ini digunakan ketika satu pipeline besar dipecah menjadi beberapa DAG terpisah, misalnya DAG pemuatan data yang menunggu DAG persiapan selesai sebelum berjalan. Parameter utamanya berupa DAG dan task yang ditunggu beserta kondisi keberhasilan yang diinginkan.

Sensor berbasis SQL menyediakan pemeriksaan langsung terhadap basis data. Provider common.sql memuat sensor SQL yang menerima sebuah query, yaitu perintah SQL yang dikirim ke basis data. Sensor ini menunggu sampai hasilnya memenuhi kondisi tertentu, misalnya baris partisi hari ini sudah muncul pada tabel sumber. Provider basis data turunan mewarisi sensor SQL tersebut, sehingga perilaku yang sama tersedia untuk mesin basis data yang didukung masing-masing provider.

Provider lain menambahkan sensor untuk sumber eksternalnya, seperti sensor berkas pada penyimpanan objek, sensor partisi pada gudang data, dan sensor pekerjaan pada sistem lainnya. Daftar lengkapnya dapat ditelusuri pada dokumentasi referensi operator dan hooks, dan pola penggunaannya selalu sama dengan FileSensor, yaitu menentukan kondisi, interval pemeriksaan, dan batas waktunya.

Kesalahan umum#

Menyamakan timeout sensor dengan keterlambatan data yang biasa terjadi. Sensor yang menunggu berkas yang datangnya rutin terlambat akan berstatus failed sebelum berkas itu tiba. Selaraskan timeout dengan latensi sumber data, atau aktifkan soft_fail=True apabila task hilir tidak wajib menunggu data tersebut.

Memilih mode poke untuk penungguan yang berlangsung lama. Sensor dalam mode poke mengunci satu slot worker selama penungguan, sehingga beberapa DAG yang menunggu bersamaan dapat menghabiskan slot untuk pekerjaan yang tidak berjalan. Gunakan mode reschedule untuk interval pemeriksaan menit atau lebih jarang, dan mode deferrable untuk penungguan yang lebih panjang.

Mengabaikan keberadaan koneksi fs_default. FileSensor yang dijalankan tanpa koneksi tersebut melempar AirflowNotFoundException saat task berjalan, bukan saat DAG diparse. Deklarasikan koneksi sebelum eksekusi, baik melalui variabel lingkungan maupun perintah airflow connections add.

Ringkasan#

• Sensor mengeksekusi fungsi poke berulang hingga mengembalikan True atau batas timeout tercapai, dan soft_fail=True mengganti status failed menjadi skipped ketika batas terlewati.
• Mode poke menempati slot worker selama penungguan, sedangkan mode reschedule melepasnya di antara pemeriksaan dengan biaya penjadwalan ulang.
• FileSensor menunggu berkas melalui koneksi fs_default dan menjadi gerbang data di hulu pipeline.
• Keluarga sensor mencakup ExternalTaskSensor, sensor SQL dari provider common.sql, dan sensor bawaan provider lainnya.

Referensi#

• Sensors — Apache Airflow — Mode poke dan reschedule serta parameter BaseSensorOperator seperti timeout, soft_fail, dan exponential_backoff.
• FileSensor — Apache Airflow Standard Provider — Parameter FileSensor untuk menunggu berkas pada sistem berkas lokal.
• ExternalTaskSensor — Apache Airflow Standard Provider — Sensor untuk menunggu selesainya task atau dag run pada DAG lain.
• Sensors — Apache Airflow Common SQL Provider — Sensor SQL yang menunggu hasil query memenuhi kondisi tertentu.
• Managing Connections — Apache Airflow — Deklarasi koneksi, termasuk koneksi fs_default dan bentuknya dari variabel lingkungan.