Dagster

Airflow mengajak kita berpikir sebagai perancang urutan langkah: task apa dijalankan, dalam urutan apa, dan kapan. Banyak masalah data justru lebih natural dinyatakan dengan pertanyaan berbeda: tabel apa, berkas apa, dan model apa yang harus dijaga kesegarannya — dan dari data apa masing-masing dihasilkan. Dagster adalah orchestrator yang dibangun sejak awal di sekitar pertanyaan tersebut lewat paradigma asset-oriented. Versi stabil saat riset course ini ditulis adalah Dagster 1.x. Pada bab ini kita membedah pergeseran paradigma dari task ke aset, menulis software-defined asset, lalu menyusun schedule dan sensor di atasnya.

Paradigma: Task versus Aset #

Perbedaan kedua paradigma paling mudah dilihat dari pertanyaan desain yang diajukan kepada penulis pipeline:

Software-defined asset adalah deklarasi objek data — tabel warehouse, berkas di object storage, hingga model machine learning — beserta komputasi yang menghasilkan objek data tersebut. Alih-alih menulis “jalankan skrip transformasi”, kita menyatakan “aset pendapatan_per_kota dihitung dari aset pesanan_bersih”. Materialization adalah peristiwa aset berhasil dihasilkan atau disegarkan pada satu titik waktu; setiap run sukses meninggalkan jejak materialization yang dapat diaudit.

Asset Graph dan Lineage Bawaan #

Kekuatan praktis paradigma aset ada pada asset graph. Dependency antaraset tidak ditulis terpisah — Dagster menyimpulkannya dari parameter fungsi: sebuah aset yang menerima pesanan_bersih sebagai argumen otomatis menjadi downstream dari pesanan_bersih. Dari informasi ini, Dagster UI (dijalankan secara lokal dengan perintah dagster dev) menggambar graf aset lengkap: lineage siapa-menghasilkan-siapa tampil tanpa alat tambahan.

Graf tersebut juga mengubah cara kita menyegarkan data. Alih-alih memicu sebuah DAG, kita memilih aset yang ingin disegarkan — lewat tombol Materialize pada UI atau CLI dagster asset materialize — dan Dagster yang menghitung aset upstream mana saja yang perlu dihitung lebih dahulu. Riwayat materialization per aset, termasuk metadata seperti jumlah baris, tersimpan untuk audit.

Menulis Software-Defined Asset #

Tiga aset berikut membentuk pipeline penjualan yang sama seperti contoh Airflow pada bab sebelumnya — perhatikan betapa sedikitnya kode yang berkaitan dengan orkestrasi:

# Sketsa software-defined asset — Dagster 1.x.
# Diverifikasi sintaks dengan py_compile; tidak dieksekusi di bab ini.

from dagster import asset


@asset(description="Pesanan mentah hasil ingestion ke zona raw.")
def pesanan_raw() -> list[dict]:
    """Baca pesanan mentah (ilustrasi: hasil job ingestion)."""
    return [
        {"id": 1, "pelanggan": "Budi", "total": 150000, "kota": "Bandung"},
        {"id": 2, "pelanggan": "Siti", "total": 75000, "kota": "Depok"},
        {"id": 3, "pelanggan": "Dewi", "total": -5000, "kota": ""},
    ]


@asset(description="Pesanan bersih: total positif dan kota terisi.")
def pesanan_bersih(pesanan_raw: list[dict]) -> list[dict]:
    return [p for p in pesanan_raw if p["total"] > 0 and p["kota"]]


@asset(description="Pendapatan per kota, dihitung dari pesanan bersih.")
def pendapatan_per_kota(pesanan_bersih: list[dict]) -> dict:
    hasil: dict = {}
    for p in pesanan_bersih:
        hasil[p["kota"]] = hasil.get(p["kota"], 0) + p["total"]
    return hasil

Mari kita telaah baris-baris kode di atas:

1. Baris 4 (import): dekorator asset adalah satu-satunya API yang dibutuhkan untuk mendefinisikan aset.
2. Baris 7-14 (pesanan_raw): dekorator @asset mengubah fungsi Python biasa menjadi aset; argumen description tampil di katalog UI sehingga pembaca pipeline memahami makna aset tanpa membuka kode.
3. Baris 17-19 (pesanan_bersih): parameter pesanan_raw sekaligus mendeklarasikan dependency — Dagster menjalankan aset upstream lebih dahulu lalu mengalirkan hasilnya sebagai argumen; baris ini juga menyaring pesanan bermasalah (total negatif, kota kosong).
4. Baris 22-27 (pendapatan_per_kota): aset agregasi di ujung graf; nilai baliknya menjadi isi materialization ketika run berhasil.

Perilaku yang diharapkan: setelah berkas ini dimuat, UI menampilkan tiga aset dengan dua sisi dependency, dan memicu materialization pendapatan_per_kota otomatis menjalankan ketiga aset secara berurutan.

Schedule dan Sensor #

Aset mendefinisikan apa yang dihasilkan; schedule dan sensor mendefinisikan kapan aset disegarkan. Schedule bekerja seperti cron pada umumnya, sedangkan sensor mengevaluasi dunia luar secara berkala — keduanya mengembalikan run request yang memicu job:

# Sketsa schedule, sensor, dan Definitions — Dagster 1.x.
# Diverifikasi sintaks dengan py_compile; tidak dieksekusi di bab ini.
# Asumsi: aset dari blok sebelumnya didefinisikan pada modul assets.py.

from pathlib import Path

from dagster import (
    Definitions,
    RunRequest,
    ScheduleDefinition,
    define_asset_job,
    sensor,
)
from assets import pendapatan_per_kota, pesanan_bersih, pesanan_raw


segarkan_semua = define_asset_job(
    name="segarkan_semua_asset", selection="*"
)

jadwal_harian = ScheduleDefinition(
    job=segarkan_semua,
    cron_schedule="0 6 * * *",
    execution_timezone="Asia/Jakarta",
)


@sensor(job=segarkan_semua, minimum_interval_seconds=30)
def sensor_berkas_baru(context):
    """Picu run saat berkas pesanan baru jatuh ke folder landing."""
    folder = Path("/data/landing/pesanan")
    semua = sorted(folder.glob("*.jsonl")) if folder.exists() else []
    kursor = context.cursor or ""
    for berkas in [b for b in semua if b.name > kursor]:
        yield RunRequest(run_key=berkas.name)
    if semua:
        context.update_cursor(semua[-1].name)


defs = Definitions(
    assets=[pesanan_raw, pesanan_bersih, pendapatan_per_kota],
    schedules=[jadwal_harian],
    sensors=[sensor_berkas_baru],
)

Empat bagian penting dari berkas tersebut:

1. Baris 17-19 (define_asset_job): membuat job yang mematerialisasi aset; selection="*" memilih seluruh aset, dan seleksi dapat dipersempit bila hanya sebagian graf yang perlu disegarkan.
2. Baris 21-25 (ScheduleDefinition): menjalankan job setiap hari pukul 06:00 zona Asia/Jakarta — perilaku setara dengan jadwal cron pada Chapter Scheduling.
3. Baris 28-37 (@sensor): sensor dievaluasi setiap minimal 30 detik. Cursor menyimpan posisi terakhir yang sudah diproses, dan run_key berdasar nama berkas mencegah satu berkas yang sama memicu dua run — pola pengaman yang menyentuh kembali konsep idempotent dari Chapter Backfill.
4. Baris 40-44 (Definitions): mendaftarkan aset, schedule, dan sensor menjadi satu titik muat yang dibaca Dagster saat startup.

Kekuatan dan Kelemahan #

Dagster juga menyediakan utilitas materialize untuk pengujian: sebuah test dapat mematerialisasi beberapa aset di proses yang sama dan memeriksa keberhasilannya, sehingga pengujian pipeline menjadi pengujian Python biasa:

# Sketsa pengujian aset dengan utilitas materialize — Dagster 1.x.
# Diverifikasi sintaks dengan py_compile; tidak dieksekusi di bab ini.

from dagster import materialize

from assets import pendapatan_per_kota, pesanan_bersih, pesanan_raw


def test_semua_aset_termaterialisasi():
    hasil = materialize([pesanan_raw, pesanan_bersih, pendapatan_per_kota])
    assert hasil.success

Dokumentasi lengkap paradigma aset, sensor, dan API lainnya tersedia di dokumentasi Dagster . Jika Dagster meminta kita berpikir sebagai aset, Prefect mengambil jalan ketiga: flow sebagai fungsi Python murni yang dinamis — kita bahas pada Chapter Prefect.