Dependency dan Task

Pipeline di dunia nyata hampir tidak pernah berupa satu skrip tunggal. Pipeline pesanan harian, misalnya, terdiri dari ekstraksi dari database sumber, validasi baris, penghitungan metrik, dan penulisan laporan — masing-masing punya risiko kegagalan dan kebutuhan sumber daya yang berbeda. Memecah pipeline menjadi task-task kecil dengan dependency eksplisit membuat setiap bagian bisa diuji, dijalankan ulang, dan dipantau secara terpisah. Chapter ini membedah anatomi task dan dependency, lalu membangun task graph mini dengan Python murni sebelum melihat wujudnya di Airflow.

Task: Satuan Kerja Terkecil #

Task adalah satu unit kerja dengan batas yang jelas: input, output, dan status sukses atau gagal. Contoh task dalam pipeline batch: membaca tabel sumber, memvalidasi schema, mengagregasi penjualan per region, dan menulis partisi harian ke warehouse. Pemecahan tersebut memberi tiga kemampuan yang tidak dimiliki skrip monolitis: retry bisa dilakukan per task tanpa mengulang seluruh pipeline, task yang tidak saling bergantung bisa berjalan paralel, dan status tiap langkah tercatat terpisah sehingga sumber masalah cepat ditemukan.

Upstream dan Downstream #

Istilah upstream menunjuk task yang harus selesai lebih dulu; downstream menunjuk task yang menunggu. “Selesai” umumnya berarti status sukses — ketika upstream gagal, downstream tidak dieksekusi dan ditandai skipped di banyak orchestrator, termasuk Airflow. Relasi upstream-downstream membentuk graf berarah: dimulai dari task tanpa upstream dan berakhir di task tanpa downstream, sementara task di tengah bisa menjadi upstream sekaligus downstream.

Tiga Tipe Dependency #

Tipe	Arti	Contoh
Data	output upstream menjadi input downstream	agregasi membaca hasil validasi
Resource	task memakai resource terbatas yang sama	dua job menulis ke tabel yang sama
Temporal	task baru boleh berjalan setelah waktu tertentu	data settlement bank tersedia pukul 05:00

Data dependency adalah tipe yang paling umum dan didukung native oleh semua orchestrator: cukup nyatakan bahwa task B menerima output task A. Resource dan temporal dependency biasanya ditangani lewat mekanisme tambahan — pool untuk membatasi jumlah task yang memakai resource sama, atau sensor yang menunggu kondisi eksternal sebelum melanjutkan.

Pola Rantai, Paralel, dan Diamond #

Tiga pola dasar muncul berulang di pipeline nyata. Pola rantai (serial) menjalankan task satu per satu: A → B → C, dengan total durasi sama dengan jumlah durasi semua task. Pola paralel memanfaatkan fan-out dan fan-in: setelah A selesai, B dan C berjalan bersamaan, dan D menunggu keduanya — total durasi mendekati durasi task terlama. Pola diamond menggabungkan keduanya dan menjadi bentuk paling umum: dua task menunggu upstream yang sama, lalu bertemu di satu task akhir.

flowchart TD A[ekstrak] --> B[validasi] B --> C[hitung total] B --> D[hitung rata-rata] C --> E[laporan] D --> E

Pada diagram di atas, hitung total dan hitung rata-rata berbagi upstream (validasi) sehingga berpotensi berjalan paralel, sementara laporan menunggu keduanya selesai.

Task Graph Mini dengan Python #

Sebelum melompat ke orchestrator sungguhan, kita bangun konsep intinya dengan Python murni: fungsi biasa berperan sebagai task, sebuah dictionary menjadi registry task, dan eksekusi manual mengikuti urutan penyisipan yang sudah topologis.

python
"""Task graph mini: dependency antar task + eksekusi manual sederhana."""


def ekstrak_pesanan():
    # simulasi baca dari database OLTP
    return [("Budi", 2), ("Siti", 1), ("Dewi", 5)]


def validasi_pesanan(pesanan):
    # buang baris dengan jumlah tidak valid
    return [p for p in pesanan if p[1] > 0]


def hitung_total(pesanan):
    return sum(jumlah for _, jumlah in pesanan)


def hitung_rata_rata(pesanan):
    return sum(jumlah for _, jumlah in pesanan) / len(pesanan)


def laporan(total, rata_rata):
    return {"total": total, "rata_rata": rata_rata}


graf = {
    "ekstrak": {"fn": ekstrak_pesanan, "upstream": []},
    "validasi": {"fn": validasi_pesanan, "upstream": ["ekstrak"]},
    "total": {"fn": hitung_total, "upstream": ["validasi"]},
    "rata_rata": {"fn": hitung_rata_rata, "upstream": ["validasi"]},
    "laporan": {"fn": laporan, "upstream": ["total", "rata_rata"]},
}


def jalankan_graf(graf):
    hasil = {}
    for nama, task in graf.items():
        for up in task["upstream"]:
            if up not in hasil:
                raise RuntimeError(f"task {nama} menunggu upstream {up}")
        argumen = [hasil[up] for up in task["upstream"]]
        hasil[nama] = task["fn"](*argumen)
        print(f"[selesai] {nama} (upstream: {', '.join(task['upstream']) or 'tidak ada'})")
    return hasil


hasil = jalankan_graf(graf)
print("laporan akhir:", hasil["laporan"])

Output yang Dihasilkan:

text
[selesai] ekstrak (upstream: tidak ada)
[selesai] validasi (upstream: ekstrak)
[selesai] total (upstream: validasi)
[selesai] rata_rata (upstream: validasi)
[selesai] laporan (upstream: total, rata_rata)
laporan akhir: {'total': 8, 'rata_rata': 2.6666666666666665}

Mari kita telaah baris-baris kode di atas:

Baris 26-32 (dictionary graf): registry menyimpan tiga hal per task — nama, fungsi, dan daftar upstream. Pola diamond terlihat jelas: total dan rata_rata sama-sama menunggu validasi, sementara laporan menunggu keduanya. Urutan penyisipan dictionary sengaja mengikuti urutan topologis.
Baris 38-40 (penegakan dependency): sebelum task dijalankan, setiap upstream diperiksa sudah memiliki hasil; bila belum, RuntimeError diangkat sehingga eksekusi manual pun tidak bisa melanggar urutan.
Baris 41-42 (data dependency): hasil upstream diteruskan sebagai argumen posisi — fungsi laporan menerima dua nilai sesuai urutan daftar upstream-nya.
Baris 43 (visibilitas mini): status setiap task dicetak saat selesai — versi paling sederhana dari yang biasanya ditampilkan orchestrator di UI-nya.

Wujud yang Sama di Airflow 3 #

Pola di atas persis yang ditulis di Airflow dengan TaskFlow API: fungsi bertanda @task, dan dependency terbentuk otomatis ketika output satu task dipakai sebagai argumen task berikutnya. Sketsa berikut mengikuti gaya Airflow 3 — import berasal dari airflow.sdk, bukan lagi airflow.decorators:

python
"""Sketsa dependency diamond dalam gaya Airflow 3 TaskFlow (konsep, tidak dieksekusi)."""
import pendulum
from airflow.sdk import dag, task


@dag(schedule=None, start_date=pendulum.datetime(2026, 1, 1, tz="UTC"), catchup=False)
def pipeline_pesanan():
    @task
    def ekstrak_pesanan():
        return [("Budi", 2), ("Siti", 1), ("Dewi", 5)]

    @task
    def validasi_pesanan(pesanan):
        return [p for p in pesanan if p[1] > 0]

    @task
    def hitung_total(pesanan):
        return sum(jumlah for _, jumlah in pesanan)

    @task
    def hitung_rata_rata(pesanan):
        return sum(jumlah for _, jumlah in pesanan) / len(pesanan)

    @task
    def laporan(total, rata_rata):
        return {"total": total, "rata_rata": rata_rata}

    pesanan = validasi_pesanan(ekstrak_pesanan())
    laporan(hitung_total(pesanan), hitung_rata_rata(pesanan))


pipeline_pesanan()

Sketsa tersebut diverifikasi sintaksnya (py_compile) tanpa dieksekusi karena membutuhkan instalasi Airflow lengkap. Detail penulisan DAG dibahas tuntas pada chapter Apache Airflow; rujukan gaya penulisan bisa dilihat di dokumentasi Apache Airflow .

Dependency menjawab pertanyaan “urutan eksekusi yang benar”. Pertanyaan berikutnya adalah pemicu: kapan dan seberapa sering seluruh graf dijalankan — pokok bahasan Chapter Scheduling.