DAG tanpa dependency hanyalah daftar task yang berjalan sendiri-sendiri. Pada bagian ini kita melengkapi dags/etl_pesanan.py dengan empat task tersisa dan menghubungkan kelimanya menjadi graf yang benar: ekstrak sebelum transform, transform sebelum muat, muat sebelum laporan, dan sebuah alarm_kegagalan yang berdiri di samping jalur utama. Airflow menyediakan dua cara mendeklarasikan dependency — implisit lewat aliran data dan eksplisit lewat operator >> — dan proyek ini sengaja memakai keduanya agar perbedaannya terlihat.

Empat Task yang Tersisa #

python dags/etl_pesanan.py — baris 68-114 
    @task(execution_timeout=timedelta(minutes=10))
    def transform(ds=None) -> dict:
        """Bersihkan dan agregasi partisi ds; kembalikan ringkasan ringan."""
        hasil = etl.baca_pesanan(RAW_DIR, ds)
        bersih = etl.bersihkan_pesanan(hasil)
        ringkasan = etl.agregasi_harian(bersih.valid, ds)
        print(
            f"[transform] {ds}: {len(bersih.valid)} valid, "
            f"{len(bersih.ditolak)} ditolak, {bersih.jumlah_duplikat} duplikat"
        )
        return {
            "ringkasan": ringkasan,
            "jumlah_baris": hasil.jumlah_baris,
            "jumlah_valid": len(bersih.valid),
            "jumlah_ditolak": len(bersih.ditolak),
            "jumlah_duplikat": bersih.jumlah_duplikat,
        }

    @task(retries=3, retry_delay=timedelta(minutes=2))
    def muat(hasil_transform: dict, ds=None) -> str:
        """Tulis ringkasan ke zona gold lalu majukan watermark di state."""
        berkas_gold = etl.muat_ringkasan(GOLD_DIR, ds, hasil_transform["ringkasan"])
        state = state_etl.baca_state(STATE_PATH)
        state = state_etl.catat_sukses(
            state,
            ds,
            jumlah_baris=hasil_transform["jumlah_baris"],
            jumlah_valid=hasil_transform["jumlah_valid"],
            jumlah_ditolak=hasil_transform["jumlah_ditolak"],
            jumlah_duplikat=hasil_transform["jumlah_duplikat"],
            jumlah_kota=len(hasil_transform["ringkasan"]),
            omzet=etl.omzet_total(hasil_transform["ringkasan"]),
            berkas_gold=str(berkas_gold),
        )
        state_etl.simpan_state(STATE_PATH, state)
        print(f"[muat] {ds}: watermark -> {state['watermark']}")
        return state["watermark"]

    @task
    def laporan(ds=None) -> None:
        """Cetak satu baris laporan hasil yang dibaca dari state."""
        print(etl.laporan_dari_state(STATE_PATH, ds))

    @task(trigger_rule="one_failed")
    def alarm_kegagalan(ds=None) -> None:
        """Stub alert: ganti print dengan notifikasi nyata di produksi."""
        print(f"[alarm] {DAG_ID} gagal memproses partisi {ds} - periksa log task")

Dua keputusan desain di sini layak digarisbawahi. Pertama, transform mengembalikan kamus kecil — ringkasan per kota plus beberapa penghitung — dan bukan seluruh baris pesanan; XCom dirancang untuk metadata berukuran kecil, sedangkan data sebesar apa pun sebaiknya mengalir lewat sistem berkas atau database. Kedua, transform membaca ulang partisi dari data/raw daripada menerima ribuan baris dari ekstrak; tukar sedikit pembacaan ulang dengan XCom yang ramping adalah pertukaran yang benar untuk pipeline ukuran ini. Detail retry pada baris 68 dan 86 dibahas penuh pada Bagian 6.

Dependency Implisit lewat XComArg #

Memanggil task TaskFlow menghasilkan XComArg — pegangan yang mewakili nilai kembalian task saat runtime. Meneruskannya sebagai argumen ke task lain sekaligus mendeklarasikan dependency:

python dags/etl_pesanan.py — baris 116-123 
    ekstrak_t = ekstrak()
    transform_t = transform()
    muat_t = muat(transform_t)  # dependency otomatis: XComArg jadi argumen
    laporan_t = laporan()
    alarm_t = alarm_kegagalan()

    ekstrak_t >> transform_t >> muat_t >> laporan_t
    [ekstrak_t, transform_t, muat_t] >> alarm_t

Baris 118 adalah dependency implisit: muat menerima transform_t, jadi Airflow tahu muat harus menunggu transform selesai dan nilainya diambil dari XCom. Perhatikan bahwa ekstrak dan transform tidak bertukar data — keduanya berbagi lewat zona data/raw — sehingga hubungan keduanya dinyatakan eksplisit.

Dependency Eksplisit dengan Operator » #

Dua baris terakhir memakai operator pergeseran kanan:

  1. Baris 122 (ekstrak_t >> transform_t >> muat_t >> laporan_t): Rantai >> berantai dari kiri ke kanan. Ekspresi a >> b di Airflow mengembalikan b, sehingga rantai ini berarti tiga sisi: ekstrak -> transform, transform -> muat, muat -> laporan.
  2. Baris 123 ([ekstrak_t, transform_t, muat_t] >> alarm_t): Operan kiri berupa daftar memasang tiga sisi sekaligus; bentuk kebalikannya (alarm_t << [ekstrak_t, transform_t, muat_t]) juga sah.

Gabungan keduanya menghasilkan graf lengkap:

Trigger Rule untuk alarm_kegagalan #

Secara bawaan sebuah task berjalan bila semua task hulunya sukses (all_success). Aturan itu tidak berguna untuk alarm: alarm justru harus berjalan ketika ada yang gagal, itulah maksud trigger_rule="one_failed" pada baris 111. Dengan aturan tersebut, alarm_kegagalan berjalan segera setelah salah satu dari ekstrak, transform, atau muat gagal, tanpa menunggu jalur utama selesai. Sebaliknya laporan tetap memakai aturan bawaan: bila muat gagal, laporan berstatus upstream_failed dan tidak dijalankan.

Wiring Dikunci oleh Pengujian ast #

Karena Airflow tidak terpasang, wiring tidak bisa diverifikasi dengan memuat DAG sungguhan. tests/test_dag_structure.py mengambil jalan yang berbeda: mengurai berkas DAG dengan modul ast bawaan Python dan menuntut keenam sisi >> ada persis seperti graf di atas:

python tests/test_dag_structure.py — baris 107-134 
def _nama(node):
    """Nama variabel pada operand operasi >> (rantai mengembalikan operan kanan)."""
    if isinstance(node, ast.Name):
        return [node.id]
    if isinstance(node, ast.List):
        return [e.id for e in node.elts if isinstance(e, ast.Name)]
    if isinstance(node, ast.BinOp) and isinstance(node.op, ast.RShift):
        return _nama(node.right)
    return []


def test_wiring_dependency_eksplisit():
    """Semua sisi >> : rantai utama plus alarm yang mengamati tiga task."""
    sisi = set()
    for node in ast.walk(_pohon()):
        if isinstance(node, ast.BinOp) and isinstance(node.op, ast.RShift):
            for hulu in _nama(node.left):
                for hilir in _nama(node.right):
                    sisi.add((hulu, hilir))
    assert sisi == {
        ("ekstrak_t", "transform_t"),
        ("transform_t", "muat_t"),
        ("muat_t", "laporan_t"),
        ("ekstrak_t", "alarm_t"),
        ("transform_t", "alarm_t"),
        ("muat_t", "alarm_t"),
    }

Fungsi _nama menelusuri nama variabel pada setiap operand >>, termasuk kasus rantai (operand kiri yang juga ekspresi >> diperlakukan sebagai operan kanannya, meniru semantik Airflow yang mengembalikan operan kanan). Pengujian ini dijalankan nyata:

bash 
python3 -m pytest tests/test_dag_structure.py -q

Output yang Dihasilkan:

text 
..........                                                               [100%]
10 passed in 0.20s

Sepuluh pengujian struktur — dari import airflow.sdk sampai instansiasi — hijau, sehingga perubahan yang tidak sengaja memutus wiring akan tertangkap sebelum menyentuh environment Airflow sungguhan. Konsep dependency dan trigger rule dibahas lebih luas pada chapter Dependency dan Task, dan daftar trigger rule lengkap ada di dokumentasi Apache Airflow .