Branching

Pipeline sering kali tidak menjalankan seluruh task pada setiap dag run, misalnya langkah pemeriksaan lanjutan yang hanya dijalankan jika nilai tertentu mencapai batas. Branching mengarahkan alur DAG dengan memilih satu atau beberapa jalur, sedangkan jalur lain diberi status skipped. Topik ini membahas decorator @task.branch, semantik skip pada task yang tidak dipilih, trigger rule untuk menyatukan jalur kembali, serta pengujian alur cabang secara lokal. Contohnya adalah DAG pemeriksaan nilai terhadap sebuah batas.

Menulis task cabang dengan @task.branch#

Decorator @task.branch bekerja seperti @task, tetapi fungsi yang didekorasi tidak mengembalikan data, melainkan nama task yang diikuti. Nilai baliknya berupa satu task_id atau daftar task_id, dan seluruh jalur lain pada hilir task cabang diberi status skipped. Nama yang dikembalikan harus merujuk task yang berada tepat di hilir task cabang, sehingga keputusan selalu diterapkan pada tahap berikutnya secara langsung.

Berkas berikut memeriksa sebuah nilai terhadap konstanta BATAS. Task pilih_cabang mengembalikan "lulus" apabila nilai mencapai batas dan "remedial" apabila tidak, kemudian kedua task tujuan bertemu kembali pada task selesai.

"""DAG yang mengarahkan alur berdasarkan nilai batas dengan @task.branch."""
import pendulum
from airflow.sdk import dag, task

BATAS = 80


@dag(
    dag_id="pemeriksaan_batas",
    schedule="@once",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
)
def pemeriksaan_batas():
    @task
    def ambil_nilai() -> int:
        return 75

    @task.branch
    def pilih_cabang(nilai: int) -> str:
        if nilai >= BATAS:
            return "lulus"
        return "remedial"

    @task
    def lulus() -> None:
        print("nilai mencapai batas kelulusan")

    @task
    def remedial() -> None:
        print("nilai di bawah batas, perlu perbaikan")

    @task(trigger_rule="none_failed_min_one_success")
    def selesai() -> None:
        print("pemeriksaan selesai")

    nilai = ambil_nilai()
    cabang = pilih_cabang(nilai)
    cabang >> [lulus(), remedial()] >> selesai()


dag = pemeriksaan_batas()

Rantai cabang >> [lulus(), remedial()] >> selesai() membentuk pola berlian yang lazim. Operator >> di sini murni mendeklarasikan dependensi eksekusi, sedangkan keputusan jalur dinyatakan oleh nilai balik pilih_cabang saat berjalan. Argumen trigger_rule pada task selesai belum dibahas pada bagian ini dan menjadi inti pembahasan berikutnya, karena tanpa pengaturan itu pola berlian tidak berjalan dengan benar.

Semantik skip pada task yang tidak dipilih#

Task yang tidak dirujuk oleh keputusan cabang diberi status skipped, dan skipped bukanlah kegagalan. Task tersebut tidak pernah dijalankan, tidak memicu retry, dan tidak menggagalkan dag run. Pada log dag run, keputusan pilih_cabang terlihat jelas. Airflow mencatat cabang yang diikuti beserta daftar task yang diberi status skipped, misalnya pesan Branch into remedial disusul Skipping tasks [('lulus', -1)] ketika nilai berada di bawah batas.

Status skip dapat merambat ke hilir melalui trigger rule. Apabila sebuah task menunggu semua hulunya sukses dengan rule bawaan all_success, satu task hulu yang skipped membuat task yang menunggu itu ikut skipped, sehingga percabangan yang tidak dirancang dengan cermat berhenti lebih awal. Aturan inilah yang menjelaskan mengapa task gabungan pada bagian berikutnya memerlukan trigger rule yang memperhitungkan task skipped.

Trigger rule untuk task gabungan#

Trigger rule menentukan kapan sebuah task layak dijadwalkan berdasarkan status task hulunya. Rule bawaan adalah all_success, yaitu seluruh task hulu harus sukses. Tabel berikut merangkum rule yang paling sering digunakan.

Pola berlian pada DAG contoh membutuhkan none_failed_min_one_success pada task selesai. Setelah percabangan, salah satu dari lulus dan remedial selalu skipped, sehingga rule bawaan all_success tidak pernah terpenuhi dan task gabungan ikut skipped. Rule none_failed_min_one_success menerima kombinasi sukses dan skipped, dengan syarat tidak ada yang gagal dan minimal satu yang sukses. Inspeksi berikut memperlihatkan trigger rule setiap task pada DAG hasil perancangan.

# memeriksa trigger rule setiap task pada DAG
python3 -c "import pemeriksaan_batas; d = pemeriksaan_batas.dag; print(d.dag_id); print({t.task_id: str(t.trigger_rule) for t in sorted(d.tasks, key=lambda t: t.task_id)})"

pemeriksaan_batas
{'ambil_nilai': 'TriggerRule.ALL_SUCCESS', 'lulus': 'TriggerRule.ALL_SUCCESS', 'pilih_cabang': 'TriggerRule.ALL_SUCCESS', 'remedial': 'TriggerRule.ALL_SUCCESS', 'selesai': 'TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS'}

Menguji alur cabang#

Pengujian menggunakan AIRFLOW_HOME terpisah, lalu berkas DAG disalin dan basis data metadata disiapkan.

# menyiapkan AIRFLOW_HOME terpisah untuk latihan topik ini
export AIRFLOW_HOME=/tmp/ra-branch
mkdir -p $AIRFLOW_HOME/dags
# menyalin berkas DAG ke folder dags
cp pemeriksaan_batas.py $AIRFLOW_HOME/dags/

# menyiapkan basis data metadata Airflow
airflow db migrate > /dev/null 2>&1
echo "kode keluar db migrate: $?"

kode keluar db migrate: 0

# menjalankan seluruh task DAG pemeriksaan_batas secara lokal
airflow dags test pemeriksaan_batas > /tmp/ra-branch/dags-test.log 2>&1
echo "kode keluar dags test: $?"

kode keluar dags test: 0

Pada log /tmp/ra-branch/dags-test.log, nilai 75 berada di bawah BATAS, sehingga pilih_cabang memilih cabang remedial. Task lulus diberi status skipped tanpa pernah berjalan. Task remedial berjalan dan sukses dengan baris log nilai di bawah batas, perlu perbaikan. Task selesai tetap berjalan karena none_failed_min_one_success terpenuhi, dan mencetak pemeriksaan selesai. Dag run berakhir dengan status successful meskipun salah satu jalurnya berstatus skipped.

Mengubah nilai balik ambil_nilai menjadi angka yang mencapai batas membalikkan keputusan pilih_cabang. Kini cabang lulus berjalan dan remedial berstatus skipped, sedangkan task selesai kembali berjalan karena pengaturan trigger rule-nya tidak berubah. Inilah ciri branching pada Airflow, yaitu struktur DAG tetap dan hanya jalur aktifnya yang berganti antar dag run.

Kesalahan umum#

Mengembalikan nama task yang bukan hilir langsung. Nilai balik pilih_cabang harus merujuk task yang berada tepat di hilir task cabang. Nama yang tidak tersambung memunculkan error saat dag run berjalan. Periksa rantai >> sebelum menuliskan nama task tujuan.

Membiarkan task gabungan menggunakan rule bawaan. Task selesai tanpa trigger_rule eksplisit menunggu all_success, padahal salah satu hulunya selalu skipped, sehingga task gabungan ikut skipped pada setiap dag run. Tetapkan none_failed_min_one_success untuk pola berlian.

Mengembalikan None tanpa sengaja. Fungsi cabang yang lalai mengembalikan nilai pada salah satu kondisi otomatis membuat semua task hilir diberi status skipped, dan dag run tampak selesai tanpa pekerjaan yang berjalan. Pastikan setiap kondisi mengembalikan nama task atau None yang memang disengaja.

Ringkasan#

• @task.branch mengembalikan satu task_id atau daftar task_id yang diikuti, dan wajib merujuk task hilir langsung dari task cabang.
• Task yang tidak dipilih berstatus skipped, bukan failed, dan skip dapat merambat melalui trigger rule all_success dan all_failed.
• Task gabungan setelah percabangan membutuhkan none_failed_min_one_success agar tetap berjalan dengan satu cabang skipped.
• Pengembalian None oleh task cabang membuat semua task yang berada tepat di hilirnya diberi status skipped.

Referensi#

• Dags — Apache Airflow Documentation — Bagian Branching dan Trigger Rules beserta contoh @task.branch dan interaksinya dengan task skipped.