Dependensi dan percabangan

Topik ini membahas cara menyusun dependensi dan percabangan antar task pada DAG, lengkap dengan aturan trigger rule, yakni aturan yang menentukan kapan sebuah task boleh dijalankan berdasarkan status task lain. Pembahasan juga mencakup sensor, yakni operator khusus yang menunggu sampai suatu kondisi terpenuhi sebelum dianggap selesai. Satu simulasi aturan trigger dijalankan dengan Python murni agar keputusannya terlihat pada tabel status.

Menyusun dependensi dengan operator »#

Dependensi pada DAG dinyatakan dengan operator pergeseran bit, yaitu >> untuk arah hulu ke hilir, sehingga kalimat a >> b dibaca b berjalan setelah a selesai. Sebuah task boleh menerima lebih dari satu syarat, dan daftar task dapat digunakan untuk menyatakan dependensi secara serentak. Pola berikut memakai kembali contoh pipeline rekap nilai ujian pada topik pembuka modul ini.

# tidak-dijalankan
baca_nilai >> gabungkan
baca_absensi >> gabungkan
gabungkan >> hitung_rata_rata >> simpan_rekap
# dua baris pertama identik dengan: [baca_nilai, baca_absensi] >> gabungkan

Pada gaya TaskFlow, dependensi tidak ditulis sebagai baris terpisah, tetapi tersusun dari pemanggilan fungsi, sebagaimana tulis_total(baca_data()) pada topik menulis DAG. Kedua bentuk tersebut menghasilkan graf yang sama, sehingga pilihannya lebih ke keterbacaan. Graf dependensi inilah yang dievaluasi scheduler pada setiap Dag run untuk menentukan task mana yang siap dijalankan.

Baris dependensi juga menjadi peta terbaik saat membaca DAG milik tim lain, karena urutan pengerjaannya terbaca tanpa membuka isi setiap fungsi. Nilai tambah yang sama terasa saat mengubah pipeline, karena menambah satu syarat berarti menambah satu baris, bukan menyunting logika fungsi yang sudah berjalan. Prinsipnya sederhana, yaitu dependensi mengurus urutan, sedangkan fungsi mengurus isi.

Percabangan dengan BranchPythonOperator#

Percabangan membiarkan keputusan diambil saat eksekusi, bukan saat DAG ditulis. BranchPythonOperator, yakni operator yang memilih cabang dengan mengembalikan task_id dari salah satu task tujuannya, menjalankan fungsi pemilih lalu menandai task cabang lain sebagai skipped.

# tidak-dijalankan
from airflow.providers.standard.operators.python import BranchPythonOperator

def pilih_cabang() -> str:
    """Memilih cabang berdasarkan banyak baris yang tersedia."""
    if baca_banyak_baris() > 1000:
        return "proses_lengkap"
    return "proses_sampel"

pilih = BranchPythonOperator(task_id="pilih_cabang", python_callable=pilih_cabang)

Gaya TaskFlow menyediakan padanannya berupa dekorator @task.branch dengan aturan kembalian yang sama, yaitu mengembalikan task_id yang dipilih. Task yang tidak dipilih tidak gagal, melainkan berstatus skipped, sehingga keputusan percabangan terekam di UI sebagai task yang dilewati dan bukan sebagai error. Percabangan layak dipakai ketika keputusannya baru diketahui saat eksekusi, misalnya ukuran data yang baru terbaca oleh task sebelumnya. Apabila keputusannya sudah pasti dari konfigurasi, cukup susun dependensi biasa dan biarkan cabang yang tidak relevan tidak dideklarasikan sama sekali.

Aturan trigger#

Aturan trigger menentukan kapan sebuah task dijalankan berdasarkan status task upstream, yakni task yang menjadi syaratnya. Aturan default adalah all_success, dan aturan lain dipilih lewat argumen trigger_rule pada task. Tabel berikut merangkum aturan yang paling sering dipakai sebagaimana didokumentasikan.

Simulasi berikut mengevaluasi lima aturan tersebut terhadap empat kombinasi status upstream yang sudah selesai, dengan Python murni tanpa mengimpor Airflow.

"""Simulasi aturan trigger Airflow terhadap status upstream yang sudah selesai."""

def all_success(status: tuple[str, ...]) -> bool:
    return all(s == "success" for s in status)


def one_success(status: tuple[str, ...]) -> bool:
    return any(s == "success" for s in status)


def one_failed(status: tuple[str, ...]) -> bool:
    return any(s == "failed" for s in status)


def all_done(status: tuple[str, ...]) -> bool:
    return all(s in ("success", "failed", "skipped") for s in status)


def none_failed_min_one_success(status: tuple[str, ...]) -> bool:
    return not any(s == "failed" for s in status) and any(s == "success" for s in status)


ATURAN = {
    "all_success": all_success,
    "one_success": one_success,
    "one_failed": one_failed,
    "all_done": all_done,
    "none_failed_min_one_success": none_failed_min_one_success,
}

skenario = {
    "A": ("success", "success"),
    "B": ("success", "failed"),
    "C": ("success", "skipped"),
    "D": ("failed", "failed"),
}

LEBAR = 30
header = f"{'aturan':<{LEBAR}}" + "".join(f"{huruf:<8}" for huruf in skenario)
print(header.rstrip())
for nama, cek in ATURAN.items():
    kolom = "".join(f"{'jalan' if cek(status) else '-':<8}" for status in skenario.values())
    print(f"{nama:<{LEBAR}}{kolom}".rstrip())

aturan                        A       B       C       D
all_success                   jalan   -       -       -
one_success                   jalan   jalan   jalan   -
one_failed                    -       jalan   -       jalan
all_done                      jalan   jalan   jalan   jalan
none_failed_min_one_success   jalan   -       jalan   -

Kolom C menunjukkan kasus setelah percabangan, yaitu satu cabang sukses dan satu cabang dilewati. Aturan all_success menolak berjalan pada kombinasi itu, sedangkan none_failed_min_one_success menerimanya karena status skipped bukan kegagalan. Kolom B memperlihatkan sifat one_failed yang bereaksi segera tanpa menunggu task upstream lain selesai, sehingga cocok untuk peringatan cepat. Kolom D menjelaskan peran all_done untuk task yang harus tetap berjalan apa pun hasilnya, misalnya pembersihan berkas sementara.

Sensor#

Sensor menyelesaikan kebutuhan yang tidak dapat dijawab dependensi biasa, yaitu menunggu sesuatu di luar DAG, misalnya kehadiran berkas atau baris data baru. Sensor memeriksa kondisinya secara berulang sampai berhasil, dengan dua mode pemakaian sumber daya berikut.

Sensor berkas berikut menunggu berkas sumber muncul, memeriksa setiap lima menit, dan gagal apabila setelah satu jam keadaan belum berubah.

# tidak-dijalankan
from airflow.providers.standard.sensors.filesystem import FileSensor

tunggu_sumber = FileSensor(
    task_id="tunggu_berkas_sumber",
    filepath="/opt/airflow/data/sumber.csv",
    poke_interval=300,
    timeout=3600,
    mode="reschedule",
)

Argumen poke_interval menetapkan jarak antar pemeriksaan, sedangkan timeout menetapkan batas waktu tunggu sebelum sensor gagal. Sensor tidak menggantikan dependensi, melainkan melengkapinya, karena hasil tunggunya tetap menjadi syarat bagi task berikutnya sebagaimana task biasa. Kombinasi keduanya menjadikan sensor cara yang tepat untuk menyambungkan penjadwalan berbasis waktu dengan kondisi dunia nyata yang datang terlambat.

Kesalahan umum#

1. Menempatkan all_success di hilir percabangan. Penyebabnya, status skipped dari cabang yang tidak terpilih menjalar ke task gabungan, sehingga task tersebut ikut terlewati. Perbaikannya, gunakan none_failed_min_one_success pada task yang menggabungkan kembali cabang-cabangnya.
2. Memakai mode poke untuk penungguan berdurasi menit. Penyebabnya, sensor tetap menahan slot worker padahal hampir sepanjang waktu tidak melakukan apa pun. Perbaikannya, pilih mode reschedule agar slot dilepas di antara pemeriksaan.
3. Menyusun dependensi yang melingkar, misalnya a bergantung b dan b bergantung a. Penyebabnya, dependensi ditambah satu per satu tanpa menggambar grafnya lebih dahulu. Perbaikannya, gambar graf dependensi sebelum menulis kode, karena DAG dengan lingkar ditolak Airflow dan tidak akan berjalan.
4. Menunggu kondisi luar dengan perulangan tidur di dalam task. Penyebabnya, kebutuhan menunggu terasa mudah dijawab dengan while dan time.sleep, padahal cara itu menahan slot worker tanpa pemantauan. Perbaikannya, pindahkan penungguan ke sensor, sehingga pemeriksaan, batas waktu, dan pemulihannya seragam di semua pipeline.

Ringkasan#

• Dependensi dinyatakan dengan operator >> pada gaya klasik, dan tersusun dari pemanggilan fungsi pada gaya TaskFlow.
• BranchPythonOperator dan @task.branch memilih cabang saat eksekusi, lalu menandai cabang lain sebagai skipped dan bukan gagal.
• Aturan trigger seperti all_success, one_failed, dan none_failed_min_one_success menentukan kapan task berjalan, dan pilihan aturan menjadi krusial di hilir percabangan.
• Sensor menunggu kondisi luar dengan mode poke atau reschedule, dikendalikan oleh poke_interval dan timeout.

Referensi#

• DAGs — Apache Airflow — Dokumentasi resmi yang memuat daftar lengkap aturan trigger dan interaksinya dengan percabangan.
• Sensors — Apache Airflow — Dokumentasi resmi konsep sensor, mode poke dan reschedule, serta parameternya.
• Python operators — Apache Airflow Standard Provider — Dokumentasi resmi BranchPythonOperator dan padanan @task.branch-nya.