Dependency Management

Tim data TokoKita kini mengoperasikan lebih dari satu pipeline: pipeline ingestion membersihkan pesanan mentah, pipeline analitik menghitung ringkasan penjualan, dan pipeline penerbitan laporan mengirim hasil ke tim bisnis. Selama urutan eksekusinya dijaga manual, semuanya berjalan; begitu ada orang yang menjalankan pipeline analitik lebih pagi dari biasanya, pipeline analitik tersebut membaca dataset setengah jadi dan menghasilkan ringkasan salah tanpa galat yang terlihat. Dependency management menutup celah ini dengan menyatakan hubungan antarpipeline secara eksplisit, sehingga urutan eksekusi dijamin benar oleh sistem, bukan oleh kebiasaan orang.

Dataset sebagai Kontrak Antarpipeline #

Cara paling sehat menyatakan dependency adalah lewat dataset, bukan lewat nama pipeline lain. Pipeline analitik tidak bergantung pada pipeline ingestion; pipeline analitik bergantung pada dataset orders_clean yang dihasilkan ingestion. Perbedaan tersebut terasa kecil, tetapi konsekuensinya besar: selama kontrak dataset terpenuhi, kita bebas menulis ulang pipeline ingestion atau menjalankannya di mesin lain tanpa menyentuh pipeline analitik.

Sebuah kontrak dataset minimal memuat empat hal: nama dataset, skema kolom, lokasi penyimpanan, dan aturan partisi. Skema berubah maka kontrak berubah, dan perubahan itu harus dinegosiasikan dengan semua konsumen sebelum diterapkan. Pola dataset-first ini juga dipakai tools orkestrasi industri; Airflow sejak versi 3 mengizinkan penjadwalan berbasis dataset sehingga scheduler menunggu dataset siap, bukan menunggu jam tertentu.

Task-Level vs Pipeline-Level #

Dependency bisa dinyatakan di dua tingkat. Tingkat task berada di dalam satu pipeline, misalnya tahap ekstraksi harus selesai sebelum tahap transformasi. Tingkat pipeline menghubungkan pipeline yang berbeda lewat dataset yang mereka produksi dan konsumsi.

Dataset Dependency vs Service Dependency #

Kategori kedua yang sering tertukar adalah sumber tunggangannya. Dataset dependency menunggu artefak data siap; service dependency menunggu sistem lain hidup dan merespons. Keduanya gagal dengan cara berbeda, sehingga penanganannya berbeda.

Siklus Dependency #

Kumpulan dependency yang sah membentuk graf berarah tanpa siklus (DAG). Siklus muncul bila pipeline A menunggu hasil B sementara B menunggu hasil A; tidak ada urutan eksekusi yang memenuhi kedua syarat tersebut, jadi siklus wajib ditolak saat pendaftaran, bukan dibiarkan sampai runtime berhenti tanpa pesan yang jelas. Pada bagian contoh, runner kita menolak konfigurasi siklus dengan pesan eksplisit sebelum satu baris pipeline pun dijalankan.

Contoh: Runner dengan Topological Sort #

Kita bangun runner kecil untuk dua pipeline TokoKita. Pipeline etl_orders membaca data/orders_2026-01-10.csv, membuang pesanan batal, dan menghasilkan dataset orders_clean berupa berkas CSV. Pipeline analytics_penjualan membaca orders_clean, menghitung ringkasan per pelanggan, dan menghasilkan dataset ringkasan_harian berupa berkas JSON. Runner mengurutkan keduanya dengan algoritma Kahn (topological sort) dan menolak siklus.

"""Runner sederhana: urutan eksekusi pipeline lewat topological sort (Kahn)."""
import csv
import json
from collections import deque
from pathlib import Path

PIPELINES = {
    "etl_orders": {"consumes": [], "produces": ["orders_clean"]},
    "analytics_penjualan": {"consumes": ["orders_clean"], "produces": ["ringkasan_harian"]},
}
OUT = Path("out")


def pipeline_etl_orders():
    with open("data/orders_2026-01-10.csv") as f:
        baris = list(csv.DictReader(f))
    valid = [b for b in baris if b["status"] != "batal"]
    OUT.mkdir(exist_ok=True)
    with (OUT / "orders_clean.csv").open("w") as f:
        f.write("order_id,pelanggan,tanggal,status,total\n")
        for b in valid:
            f.write(f"{b['order_id']},{b['pelanggan']},{b['tanggal']},"
                    f"{b['status'].lower()},{b['total']}\n")
    print(f"etl_orders: {len(baris)} dibaca, {len(valid)} valid -> {OUT / 'orders_clean.csv'}")


def pipeline_analytics_penjualan():
    with (OUT / "orders_clean.csv").open() as f:
        valid = list(csv.DictReader(f))
    ringkasan = {}
    for b in valid:
        nama = b["pelanggan"]
        jumlah, total = ringkasan.get(nama, (0, 0))
        ringkasan[nama] = (jumlah + 1, total + int(b["total"]))
    (OUT / "ringkasan_harian.json").write_text(
        json.dumps(ringkasan, indent=2, sort_keys=True) + "\n")
    print(f"analytics_penjualan: {len(ringkasan)} pelanggan -> "
          f"{OUT / 'ringkasan_harian.json'}")
    for nama, (jumlah, total) in sorted(ringkasan.items()):
        print(f"  {nama}: {jumlah} pesanan, Rp{total}")


FUNGSI = {"etl_orders": pipeline_etl_orders,
          "analytics_penjualan": pipeline_analytics_penjualan}


def urutan_eksekusi(pipelines):
    produsen = {ds: nama for nama, info in pipelines.items() for ds in info["produces"]}
    for nama, info in pipelines.items():
        for ds in info["consumes"]:
            if ds not in produsen:
                raise ValueError(f"dataset {ds!r} tidak memiliki produsen")
    derajat = {nama: 0 for nama in pipelines}
    tetangga = {nama: [] for nama in pipelines}
    for nama, info in pipelines.items():
        for ds in info["consumes"]:
            derajat[nama] += 1
            tetangga[produsen[ds]].append(nama)
    antrean = deque(sorted(n for n, d in derajat.items() if d == 0))
    urutan = []
    while antrean:
        nama = antrean.popleft()
        urutan.append(nama)
        for berikutnya in sorted(tetangga[nama]):
            derajat[berikutnya] -= 1
            if derajat[berikutnya] == 0:
                antrean.append(berikutnya)
    if len(urutan) != len(pipelines):
        tersisa = sorted(set(pipelines) - set(urutan))
        raise ValueError(f"dependency cycle melibatkan: {', '.join(tersisa)}")
    return urutan


print("=== DAG normal: dua pipeline ===")
for nama in urutan_eksekusi(PIPELINES):
    print(f"jalankan {nama}")
    FUNGSI[nama]()

print("=== DAG dengan siklus ===")
pipelines_siklus = dict(PIPELINES)
pipelines_siklus["etl_orders"] = {"consumes": ["ringkasan_harian"],
                                  "produces": ["orders_clean"]}
try:
    urutan_eksekusi(pipelines_siklus)
except ValueError as galat:
    print(f"ditolak saat registrasi: {galat}")

Output yang Dihasilkan:

=== DAG normal: dua pipeline ===
jalankan etl_orders
etl_orders: 5 dibaca, 4 valid -> out/orders_clean.csv
jalankan analytics_penjualan
analytics_penjualan: 3 pelanggan -> out/ringkasan_harian.json
  Budi: 2 pesanan, Rp1100000
  Dewi: 1 pesanan, Rp275000
  Siti: 1 pesanan, Rp150000
=== DAG dengan siklus ===
ditolak saat registrasi: dependency cycle melibatkan: analytics_penjualan, etl_orders

Mari kita telaah baris-per-baris kode di atas:

1. Baris 7-10 (registrasi): setiap pipeline mendaftarkan dua daftar, yaitu dataset yang dikonsumsi dan dataset yang diproduksi; informasi inilah yang menjadi sumber kebenaran urutan eksekusi.
2. Baris 48: dictionary produsen memetakan setiap dataset ke pipeline penghasilnya, sekaligus menjadi tabel pencarian saat memvalidasi konsumsi.
3. Baris 49-52: konsumsi terhadap dataset tanpa produsen langsung ditolak, sehingga salah ketik nama dataset terungkap saat registrasi.
4. Baris 53-58: tabel derajat menghitung berapa dependency yang belum terpenuhi per pipeline, dan tetangga mencatat pipeline mana yang aktif begitu sebuah pipeline selesai.
5. Baris 59-67 (algoritma Kahn): pipeline dengan derajat nol masuk antrean, dikeluarkan satu per satu, lalu derajat tetangganya diturunkan; pipeline baru yang derajatnya mencapai nol ikut masuk antrean.
6. Baris 68-70 (deteksi siklus): bila urutan hasil lebih pendek dari jumlah pipeline, berarti ada pipeline yang derajatnya tidak pernah mencapai nol, dan daftarnya dilaporkan sebagai siklus.

Percobaan kedua pada keluaran sengaja memaksa siklus: etl_orders dibuat ikut mengonsumsi ringkasan_harian, sehingga kedua pipeline saling menunggu. Runner menolak konfigurasi tersebut sebelum mengeksekusi apa pun, dengan pesan yang menyebut kedua anggota siklus.

Penjadwalan berbasis dataset dan deteksi siklus pada skala produksi ditangani tools orkestrasi; rujukan resminya ada di dokumentasi Apache Airflow untuk konsep DAG dan dataset, serta dokumentasi dbt untuk dependency antarmodel transformasi. Urutan eksekusi kini terjamin, tetapi urutan yang benar belum menutup satu risiko: pipeline yang dijalankan ulang setelah gagal. Pada Chapter Idempotency dan Retries kita memastikan eksekusi ulang tidak merusak data.