KAL
Menu

Arsitektur DAG

Memecah platform data menjadi beberapa DAG dengan batas tanggung jawab yang jelas.

Daftar isi

Platform data statistik ujian dari topik pemilihan proyek kini dirancang menjadi arsitektur Airflow. Topik ini membahas prinsip dekomposisi DAG, batas tanggung jawab ingesta, olah, dan ekspor, serta penghubung antar DAG, lalu menerapkannya pada rancangan tiga DAG untuk proyek puncak. Dua DAG pertama ditulis lengkap dan diperiksa strukturnya, sedangkan DAG ingesta dijalankan penuh dengan perintah dags test, sehingga rancangannya dapat langsung diuji.

Prinsip dekomposisi DAG#

Dekomposisi adalah pemecahan satu platform data menjadi beberapa DAG yang masing-masing bertanggung jawab atas satu pekerjaan. Alasan pertama bersifat teknis, yaitu DAG merupakan satuan penjadwalan yang hanya membawa satu definisi jadwal. Platform statistik ujian memuat jadwal yang berbeda antartahap. Ingesta berjalan setiap pukul 06.00 melalui cron, sedangkan olah data seharusnya berjalan sesegera mungkin setelah ingesta selesai dan tidak menunggu jadwal apa pun. Menyatukan keduanya memaksa salah satu tahap mengikuti jadwal yang salah.

Alasan kedua bersifat operasional karena DAG juga merupakan satuan pemantauan dan pemulihan. Tahap ingesta, olah, dan ekspor memiliki profil kegagalan yang berbeda. Ingesta gagal apabila sumbernya belum tersedia, sedangkan tahap olah gagal apabila data tidak memenuhi aturan mutu. Dengan batas tanggung jawab yang jelas, penjadwalan ulang, backfill, dan penanganan error dilakukan per tahap tanpa menyentuh tahap lainnya. Dalam satu DAG yang jadwalnya memang sama, pengelompokan visual cukup dilakukan dengan task group karena seluruh task di dalamnya selalu berjalan bersama.

Kebalikan prinsip ini adalah DAG monolitik, yaitu satu DAG yang menampung seluruh tahap platform dan memaksakan satu jadwal untuk semuanya. Bentuk ini lazim muncul ketika platform dibangun tanpa rancangan terlebih dahulu, lalu tumbuh menjadi satu berkas yang panjang.

Penghubung antar DAG#

Pemisahan DAG menimbulkan kebutuhan baru, yaitu menghubungkan dag run pada satu DAG dengan dag run pada DAG lain. Dua mekanisme paling umum digunakan sebagai penghubung. TriggerDagRunOperator adalah task yang memicu dag run pada DAG lain setelah task di atasnya selesai, sedangkan ExternalTaskSensor adalah sensor yang menunggu task atau dag run pada DAG lain mencapai status tertentu sebelum melanjutkan alur. Keduanya memindahkan kebergantungan antartahap yang semula berupa urutan task menjadi kebergantungan antar DAG.

Airflow 3 juga menyediakan penghubung berbasis data yang disebut asset, yaitu nama baru untuk konsep dataset pada versi sebelumnya. Task hulu memperbarui sebuah asset, sedangkan DAG hilir berjalan secara otomatis ketika asset tersebut diperbarui. Mekanisme ini berguna apabila hubungan antar DAG mengikuti kesiapan data dan bukan urutan pekerjaan. Untuk proyek kecil dengan tahap yang jelas urutannya, pemicu eksplisit lebih mudah dipahami dan didebug, sehingga proyek statistik ujian menggunakan pemicu antar DAG.

Rancangan tiga DAG proyek statistik ujian#

Diagram berikut memperlihatkan pembagian tiga DAG pada proyek beserta aliran datanya, dari jadwal harian di paling atas hingga ekspor di paling bawah.

Pembagian tiga DAG pada proyek statistik ujian sekolahdag_ingesta_harianjadwal: 0 6 * * *salin_nilaisalin_kehadiransalin_suhudipicu setelah ingesta selesaidag_olah_nilaitanpa jadwalbaca_ingestaperiksa_mutuagregat_nilaidipicu setelah olah nilai selesaidag_ringkasan_eksportanpa jadwalsusun_ringkasanekspor_ringkasan
Pembagian tiga DAG pada proyek statistik ujian sekolah

DAG pertama, dag_ingesta_harian, berjadwal 0 6 * * * dan memuat tiga task yang hanya menyalin nilai, kehadiran, serta catatan suhu ke folder ingesta yang dipartisi per tanggal, tanpa memahami isi datanya sama sekali. DAG kedua, dag_olah_nilai, tanpa jadwal dan memuat tiga task, yaitu membaca hasil ingesta, menjalankan gerbang mutu, lalu mengagregasi nilai dan kehadiran per kelas. DAG ini hanya membaca folder ingesta dan tidak mengetahui dari mana CSV mentah berasal. DAG ketiga, dag_ringkasan_ekspor, juga tanpa jadwal dan bertugas menyusun ringkasan lintas kelas lalu mengekspornya. Task pada DAG ketiga diisi lengkap pada topik implementasi, sedangkan dua DAG pertama ditulis berikut ini.

Berkas pertama memuat DAG ingesta harian. Ketiga task salin berjalan tanpa saling bergantung karena sumbernya berbeda, dan setiap task menuliskan hasilnya ke folder yang dipartisi ds_nodash, yaitu tanggal data interval tanpa tanda hubung. Pemanggilan os.makedirs dengan exist_ok=True membuat penulisan ulang pada tanggal yang sama aman, sehingga setiap task idempoten.

python dag_ingesta_harian.py 
"""DAG ingesta harian proyek statistik ujian sekolah.

Menyalin tiga sumber CSV mentah (nilai, kehadiran, catatan suhu ruang ujian)
ke folder ingesta yang dipartisi per tanggal.
"""
import os

import pendulum
from airflow.sdk import dag, task

FOLDER_INGESTA = "/tmp/ra-statistik/ingesta"

NILAI_CSV = """nis,nama,kelas,nilai
1001,Ali,A,82
1002,Budi,A,74
1003,Citra,A,91
1004,Dewi,B,66
1005,Eka,B,78
1006,Fani,B,85
"""

KEHADIRAN_CSV = """nis,tanggal,status
1001,2026-11-04,hadir
1002,2026-11-04,hadir
1003,2026-11-04,izin
1004,2026-11-04,hadir
1005,2026-11-04,hadir
1006,2026-11-04,sakit
"""

SUHU_CSV = """ruang,waktu,suhu_c
R-101,08:00,25.1
R-101,10:00,26.3
R-102,08:00,24.8
R-102,10:00,25.6
"""


@dag(
    dag_id="dag_ingesta_harian",
    schedule="0 6 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    description="Menyalin tiga sumber CSV mentah ke folder ingesta harian.",
)
def dag_ingesta_harian():
    @task
    def salin_nilai(ds_nodash: str) -> str:
        """Menyalin berkas nilai mentah ke folder ingesta tanggal ini."""
        folder = f"{FOLDER_INGESTA}/{ds_nodash}"
        os.makedirs(folder, exist_ok=True)
        tujuan = f"{folder}/nilai.csv"
        with open(tujuan, "w", encoding="utf-8") as berkas:
            berkas.write(NILAI_CSV)
        return tujuan

    @task
    def salin_kehadiran(ds_nodash: str) -> str:
        """Menyalin berkas kehadiran mentah ke folder ingesta tanggal ini."""
        folder = f"{FOLDER_INGESTA}/{ds_nodash}"
        os.makedirs(folder, exist_ok=True)
        tujuan = f"{folder}/kehadiran.csv"
        with open(tujuan, "w", encoding="utf-8") as berkas:
            berkas.write(KEHADIRAN_CSV)
        return tujuan

    @task
    def salin_suhu(ds_nodash: str) -> str:
        """Menyalin catatan suhu ruang ujian ke folder ingesta tanggal ini."""
        folder = f"{FOLDER_INGESTA}/{ds_nodash}"
        os.makedirs(folder, exist_ok=True)
        tujuan = f"{folder}/suhu.csv"
        with open(tujuan, "w", encoding="utf-8") as berkas:
            berkas.write(SUHU_CSV)
        return tujuan

    salin_nilai()
    salin_kehadiran()
    salin_suhu()


dag_ingesta_harian()

Berkas kedua memuat DAG olah nilai. Nilai schedule=None menegaskan bahwa DAG ini tidak memiliki jadwal sendiri. Task periksa_mutu menjadi gerbang mutu karena memunculkan RuntimeError saat ada baris yang melanggar aturan, sehingga task agregasi di hilir tidak pernah menerima data yang tidak sah.

python dag_olah_nilai.py 
"""DAG olah nilai proyek statistik ujian sekolah.

Membaca hasil ingesta harian, menjalankan gerbang mutu, lalu menghitung
agregasi nilai dan kehadiran per kelas. DAG ini tidak memiliki jadwal karena
dipicu setelah DAG ingesta harian menuntaskan pekerjaannya.
"""
from airflow.sdk import dag, task

FOLDER_INGESTA = "/tmp/ra-statistik/ingesta"


def baca_csv(path: str) -> list[dict]:
    """Membaca satu berkas CSV kecil menjadi list of dict."""
    with open(path, encoding="utf-8") as berkas:
        baris = berkas.read().strip().split("\n")
    kepala = baris[0].split(",")
    return [dict(zip(kepala, b.split(","))) for b in baris[1:]]


@dag(
    dag_id="dag_olah_nilai",
    schedule=None,
    description="Gerbang mutu lalu agregasi nilai dan kehadiran per kelas.",
)
def dag_olah_nilai():
    @task
    def baca_ingesta(ds_nodash: str) -> dict:
        """Membaca berkas nilai dan kehadiran dari folder ingesta hari ini."""
        folder = f"{FOLDER_INGESTA}/{ds_nodash}"
        return {
            "nilai": baca_csv(f"{folder}/nilai.csv"),
            "kehadiran": baca_csv(f"{folder}/kehadiran.csv"),
        }

    @task
    def periksa_mutu(data: dict) -> dict:
        """Menahan alur bila ada nilai kosong atau di luar rentang 0 sampai 100."""
        data["nilai"] = [
            {**b, "nilai": int(b["nilai"])}
            for b in data["nilai"]
            if b["nilai"] != ""
        ]
        pelanggaran = [
            b["nis"] for b in data["nilai"] if not 0 <= b["nilai"] <= 100
        ]
        if pelanggaran:
            raise RuntimeError(f"gerbang mutu gagal: {len(pelanggaran)} baris tidak memenuhi aturan")
        print(f"gerbang mutu lolos untuk {len(data['nilai'])} baris")
        return data

    @task
    def agregat_nilai(data: dict) -> dict:
        """Menghitung rata-rata nilai dan tingkat kehadiran per kelas."""
        kelas = sorted({b["kelas"] for b in data["nilai"]})
        hasil = {}
        for k in kelas:
            baris = [b for b in data["nilai"] if b["kelas"] == k]
            nis = {str(b["nis"]) for b in baris}
            hadir = sum(
                1 for b in data["kehadiran"] if b["nis"] in nis and b["status"] == "hadir"
            )
            hasil[k] = {
                "rata_nilai": round(sum(b["nilai"] for b in baris) / len(baris), 1),
                "persen_hadir": round(100 * hadir / len(nis)),
            }
        print(hasil)
        return hasil

    data = baca_ingesta()
    sah = periksa_mutu(data)
    agregat_nilai(sah)


dag_olah_nilai()

Memeriksa struktur dua DAG#

Sebelum DAG dipasang, strukturnya diperiksa terlebih dahulu. Variabel lingkungan menetapkan folder latihan terpisah, lalu kedua berkas dijalankan langsung dengan Python untuk menguji parse senyap, yaitu berkas diparse tanpa keluaran dan tanpa pesan error.

bash 
# lingkungan latihan terpisah agar tidak bercampur dengan instalasi lain
export AIRFLOW_HOME=/tmp/ra-arsitektur-dag
# contoh DAG bawaan dimatikan supaya pemrosesan lebih cepat
export AIRFLOW__CORE__LOAD_EXAMPLES=False
bash 
# uji parse senyap kedua berkas DAG
python3 dag_ingesta_harian.py
python3 dag_olah_nilai.py

Pemeriksaan struktur berikut memuat modul DAG, memanggil fungsi DAG-nya, lalu mencetak dag_id, jadwal, dan seluruh task_id dalam urutan alfabetis.

bash 
# periksa dag_id, jadwal, dan daftar task DAG ingesta
python3 -c "import dag_ingesta_harian; d = dag_ingesta_harian.dag_ingesta_harian(); print(d.dag_id); print(d.schedule); print(sorted(t.task_id for t in d.tasks))"
text Keluaran 
dag_ingesta_harian
0 6 * * *
['salin_kehadiran', 'salin_nilai', 'salin_suhu']
bash 
# periksa dag_id, jadwal, dan daftar task DAG olah nilai
python3 -c "import dag_olah_nilai; d = dag_olah_nilai.dag_olah_nilai(); print(d.dag_id); print(d.schedule); print(sorted(t.task_id for t in d.tasks))"
text Keluaran 
dag_olah_nilai
None
['agregat_nilai', 'baca_ingesta', 'periksa_mutu']

Kedua keluaran pemeriksaan menegaskan rancangan pada diagram. DAG ingesta terjadwal 0 6 * * * dengan tiga task salin, sedangkan DAG olah nilai bernilai None pada jadwalnya dan memuat tiga task olah. Nilai None pada baris kedua bukan kejanggalan, melainkan bentuk internal dari schedule=None.

Menjalankan DAG ingesta#

DAG ingesta kini dijalankan penuh. Kedua berkas DAG disalin ke folder dags milik AIRFLOW_HOME, basis data metadata disiapkan dengan db migrate, lalu dags test menjalankan satu dag run secara lokal tanpa scheduler maupun worker terpisah.

bash 
# siapkan folder dags lalu salin kedua berkas DAG ke dalamnya
export AIRFLOW_HOME=/tmp/ra-arsitektur-dag
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_ingesta_harian.py dag_olah_nilai.py "$AIRFLOW_HOME/dags/"
bash 
# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
export AIRFLOW_HOME=/tmp/ra-arsitektur-dag
airflow db migrate > /dev/null 2>&1
bash 
# jalankan seluruh task DAG ingesta sekali secara lokal
# keluaran penuh dialihkan ke berkas log karena bertanda waktu
export AIRFLOW_HOME=/tmp/ra-arsitektur-dag
export AIRFLOW__CORE__LOAD_EXAMPLES=False
airflow dags test dag_ingesta_harian 2026-11-04 > /tmp/ra-arsitektur-dag-ingesta.log 2>&1

Perintah terakhir berjalan sekitar sepuluh detik dan berakhir tanpa pesan apa pun. Pada bagian akhir log, ketiga task salin berstatus success dan dag run ditutup dengan status yang sama. Bukti keberhasilannya diambil dari log dengan menghitung baris status, lalu dari folder ingesta yang merupakan hasil kerja sesungguhnya dari task.

bash 
# hitung task yang berakhir success pada log
grep -c "new_state=success" /tmp/ra-arsitektur-dag-ingesta.log
# pastikan dag run ditutup dengan status success
grep -c "DagRun Finished: dag_id=dag_ingesta_harian" /tmp/ra-arsitektur-dag-ingesta.log
text Keluaran 
3
1
bash 
# tampilkan isi berkas nilai hasil ingesta tanggal 2026-11-04
cat /tmp/ra-statistik/ingesta/20261104/nilai.csv
text Keluaran 
nis,nama,kelas,nilai
1001,Ali,A,82
1002,Budi,A,74
1003,Citra,A,91
1004,Dewi,B,66
1005,Eka,B,78
1006,Fani,B,85

Angka 3 pada keluaran pertama menyatakan tiga task salin yang berstatus success, sedangkan angka 1 menyatakan satu baris penutup dag run yang berstatus success. Berkas nilai.csv pada folder 20261104 menunjukkan bahwa partisi tanggal terbentuk sesuai ds_nodash dari data interval dag run. Hasil yang sama berlaku untuk kehadiran.csv dan suhu.csv.

Ringkasan#

  • Dekomposisi memecah platform menjadi beberapa DAG dengan batas satu jadwal dan satu tanggung jawab per DAG, sedangkan task group hanya digunakan di dalam DAG yang jadwalnya memang sama.
  • Penghubung antar DAG berupa TriggerDagRunOperator, ExternalTaskSensor, atau asset yang pada Airflow 3 menggantikan konsep dataset.
  • Proyek statistik ujian terbagi menjadi dag_ingesta_harian berjadwal cron, dag_olah_nilai tanpa jadwal, dan dag_ringkasan_ekspor tanpa jadwal.
  • dags test menjalankan DAG ingesta penuh, dan bukti keberhasilannya dibaca dari hitungan status pada log serta isi folder ingesta.

Referensi#