KAL
Menu

Dependensi

Mengatur dependensi antar task dengan bitshift dan list, termasuk pola fan-out dan fan-in.

Daftar isi

Dependensi menyatakan urutan pelaksanaan antar task di dalam DAG, dan Airflow menyediakan notasi panah untuk menuliskannya secara ringkas. Topik ini membahas notasi bitshift, bentuk daftar untuk menghubungkan satu task dengan banyak task, metode eksplisit set_downstream dan set_upstream, serta pola fan-out dan fan-in yang membentuk graf berlian. Graf yang dihasilkan juga dibaca ulang lewat pemeriksaan Python dan tampilan Graph view agar hubungan antar task dapat diverifikasi dengan pasti.

Istilah upstream dan downstream#

Task yang harus selesai lebih dulu disebut upstream, sedangkan task yang menunggu hasilnya disebut downstream. Istilah ini selalu relatif terhadap sebuah task tertentu, sehingga satu task dapat menjadi downstream terhadap sebuah task dan sekaligus menjadi upstream terhadap task yang lain. Hubungan yang dimaksud adalah hubungan langsung, yaitu tersambung lewat satu panah, bukan seluruh task yang berada lebih awal pada graf.

Aturan bawaan Airflow menentukan bahwa task baru berjalan apabila seluruh task upstream yang tersambung langsung dengannya berstatus success. Dengan aturan ini, kegagalan pada satu task menghentikan seluruh task hilirnya secara otomatis, tanpa memerlukan kode tambahan. Aturan ini dapat diubah dengan trigger rule, yaitu aturan yang menentukan kapan task boleh berjalan, untuk kebutuhan percabangan. Contoh pada topik ini cukup mengandalkan aturan bawaan tersebut.

Panah dependensi hanya mengatur urutan pelaksanaan, bukan aliran data. Task hilir tidak menerima nilai dari task hulunya hanya karena tersambung panah. Pengoperan data kecil antar task memerlukan XCom, yaitu penyimpanan nilai sementara yang dapat dibaca task lain. Pada DAG yang dibangun di bagian selanjutnya, deret nilai yang diproses setiap task ditulis langsung di dalam fungsinya karena pembahasan masih berfokus pada bentuk grafnya.

Panah bitshift dan metode eksplisit#

Dependensi dideklarasikan dengan operator bitshift >> dan <<. Ekspresi a >> b menyatakan bahwa task b merupakan downstream dari task a, sedangkan a << b menyatakan hal yang sama dengan arah pembacaan terbalik. Bentuk daftar menghubungkan satu task dengan banyak task sekaligus, misalnya a >> [b, c] yang menjadikan b dan c downstream dari a. Daftar tersebut dapat dirantai, sehingga pola berlian utuh ditulis dalam satu baris a >> [b, c] >> d.

Airflow juga menyediakan metode set_downstream dan set_upstream yang bekerja sama persis dengan panah bitshift. Kedua bentuk berikut menyatakan hubungan yang identik dengan baris siapkan >> [hitung_jumlah, hitung_maksimum] >> laporan pada DAG yang akan dibangun.

python Contoh: metode set_downstream dan set_upstream 
# tidak-dijalankan
siapkan.set_downstream([hitung_jumlah, hitung_maksimum])
laporan.set_upstream([hitung_jumlah, hitung_maksimum])

Dokumentasi resmi menyarankan panah bitshift karena lebih mudah dibaca pada sebagian besar kasus, sedangkan metode eksplisit berguna ketika dependensi disusun di dalam perulangan atau fungsi pembantu. Satu batasan yang perlu dijaga adalah tidak terbentuknya siklus. Apabila task saling menunjuk hingga kembali ke titik awal, berkas DAG gagal diparse dengan pesan error bahwa siklus terdeteksi, sehingga DAG tidak termuat sama sekali.

Penolakan siklus tersebut dapat diperagakan dengan pemeriksaan sederhana di luar Airflow. Graf dependensi dimodelkan sebagai kamus berupa pemetaan dari satu task ke daftar task hilirnya. Penelusuran mendalam (depth-first search) menandai task yang sedang dikunjungi dan melaporkan siklus begitu penelusuran kembali ke task yang masih ditandai. Skrip berikut membandingkan graf berlian tanpa panah balik dengan graf yang sama setelah satu panah balik ditambahkan.

python Contoh: deteksi siklus pada graf dependensi 
graf_berlian = {
    "siapkan": ["jumlah", "maksimum"],
    "jumlah": ["laporan"],
    "maksimum": ["laporan"],
    "laporan": [],
}

# panah balik dari laporan ke siapkan membentuk siklus
graf_siklus = {nama: list(hilir) for nama, hilir in graf_berlian.items()}
graf_siklus["laporan"].append("siapkan")


def ada_siklus(graf):
    """Menelusuri graf dan melaporkan keberadaan siklus."""
    status = {nama: 0 for nama in graf}

    def kunjungi(nama):
        if status[nama] == 1:
            return True
        if status[nama] == 2:
            return False
        status[nama] = 1
        ketemu = any(kunjungi(hilir) for hilir in graf[nama])
        status[nama] = 2
        return ketemu

    return any(kunjungi(nama) for nama in graf)


print(f"Berlian tanpa panah balik: {ada_siklus(graf_berlian)}")
print(f"Berlian dengan panah balik: {ada_siklus(graf_siklus)}")
text Keluaran 
Berlian tanpa panah balik: False
Berlian dengan panah balik: True

Keluaran menunjukkan graf berlian tanpa panah balik dinyatakan bebas siklus, sedangkan penambahan satu panah balik membuat penelusuran kembali ke task yang masih ditandai. Airflow menjalankan pemeriksaan yang setara ketika berkas DAG diparse, sehingga siklus yang tidak sengaja terbentuk langsung dilaporkan sebagai error dan DAG tidak dimuat sama sekali.

Pola fan-out dan fan-in#

Fan-out adalah pola saat satu task membagi pekerjaan ke banyak task hilir, sedangkan fan-in adalah pola saat banyak task berkumpul ke satu task hilir. Gabungan kedua pola menghasilkan graf berlian (diamond), yaitu bentuk dependensi yang paling sering muncul pada pipeline data karena memisahkan tahap persiapan, perhitungan paralel, dan penyimpanan hasil. Airflow menjalankan task pada cabang yang berbeda secara paralel selama slot eksekusi tersedia, sehingga perhitungan pada kedua cabang tidak saling menunggu.

Berkas dag_proses.py berikut membangun graf berlian dengan empat task gaya klasik. Task siapkan_data menjadi sumber, dua task perhitungan menjadi cabang tengah, dan simpan_laporan menjadi titik kumpulnya.

python dag_proses.py 
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import dag


def hitung_jumlah() -> None:
    """Menghitung jumlah dari deret nilai yang disiapkan task pertama."""
    angka = [3, 7, 2, 9, 4]
    print(f"Jumlah: {sum(angka)}")


def hitung_maksimum() -> None:
    """Menghitung nilai terbesar dari deret nilai yang sama."""
    angka = [3, 7, 2, 9, 4]
    print(f"Maksimum: {max(angka)}")


@dag(
    dag_id="dag_proses",
    schedule=None,
    description="Pola berlian: satu sumber, dua cabang paralel, satu gabungan.",
)
def dag_proses():
    siapkan = BashOperator(
        task_id="siapkan_data",
        bash_command="echo 'Data siap: 3, 7, 2, 9, 4'",
    )
    jumlah = PythonOperator(
        task_id="hitung_jumlah",
        python_callable=hitung_jumlah,
    )
    maksimum = PythonOperator(
        task_id="hitung_maksimum",
        python_callable=hitung_maksimum,
    )
    laporan = BashOperator(
        task_id="simpan_laporan",
        bash_command="echo 'Laporan selesai'",
    )

    siapkan >> [jumlah, maksimum] >> laporan


dag_proses()

Struktur dependensinya membentuk graf berlian seperti pada diagram berikut.

siapkan_datahitung_jumlahhitung_maksimumsimpan_laporan
Graf berlian pada dag_proses dengan satu sumber, dua cabang paralel, dan satu gabungan

Hubungan antar task yang tertulis sebagai panah dapat ditanyakan kembali dari objek DAG yang telah diparse. Properti downstream_task_ids dan upstream_task_ids milik tiap task menampung nama task yang tersambung langsung dengannya.

bash 
# muat DAG lalu cetak task hilir siapkan_data dan task hulu simpan_laporan
python3 -c "import dag_proses; d = dag_proses.dag_proses(); print(d.dag_id); print(sorted(d.get_task('siapkan_data').downstream_task_ids)); print(sorted(d.get_task('simpan_laporan').upstream_task_ids))"
text Keluaran 
dag_proses
['hitung_jumlah', 'hitung_maksimum']
['hitung_jumlah', 'hitung_maksimum']

Keluaran memuat dua daftar yang isinya sama, dan kesamaan itu justru menegaskan bentuk berliannya. Task siapkan_data memiliki dua task hilir, sedangkan task simpan_laporan memiliki dua task hulu yang sama persis. Baris siapkan >> [jumlah, maksimum] >> laporan karenanya terbukti membentuk satu fan-out dan satu fan-in sekaligus.

Membaca graf pada Graph view#

Graph view pada UI menampilkan graf dependensi persis seperti diagram di atas, yaitu setiap task menjadi satu simpul dan setiap panah menjadi satu garis penghubung. Graf tersebut dibaca dari kiri ke kanan mengikuti arah panah, dan tampilannya sama dengan hasil pemeriksaan Python karena keduanya bersumber dari data dependensi yang sama. Apabila cabang tengah bertambah, Graph view akan menampilkan cabang baru tanpa perubahan apa pun pada task lain, sehingga bentuk berliannya tetap terlihat jelas. Mengeklik satu simpul membuka detail task beserta status terkininya, sehingga kegagalan pada satu cabang dapat dilacak tanpa kehilangan konteks graf keseluruhan.

Ringkasan#

  • Dependensi ditulis dengan >> atau <<, dan bentuk daftar seperti a >> [b, c] >> d menyusun fan-out sekaligus fan-in dalam satu baris.
  • Metode set_downstream dan set_upstream menghasilkan graf yang sama dengan panah bitshift, tetapi panah bitshift lebih mudah dibaca.
  • Task baru berjalan apabila seluruh upstream langsungnya success, sedangkan DAG yang mengandung siklus ditolak saat diparse.
  • Properti downstream_task_ids dan upstream_task_ids serta Graph view membantu memverifikasi graf tanpa menjalankan task.

Referensi#