KAL
Menu

Implementasi

Mengimplementasikan pipeline proyek dengan TaskFlow, sensor, dan gerbang mutu yang terintegrasi.

Daftar isi

Rancangan tiga DAG pada proyek statistik ujian menyisakan dag_ringkasan_ekspor, yang pada topik arsitektur masih berupa dua kotak pada diagram. Topik ini mengisi DAG tersebut dengan tugas menyusun ringkasan lintas kelas dan mengekspornya, lalu menambahkan dua penjaga produksi berupa sensor yang menunggu data ingesta dan gerbang mutu yang menahan alur. Seluruh DAG dieksekusi dua kali, pertama dengan data sah yang berhasil diekspor dan kedua dengan data cacat yang ditolak gerbang, sehingga setiap pernyataan pada topik ini punya bukti keluarannya sendiri.

Menerjemahkan diagram menjadi DAG#

Diagram arsitektur menampilkan DAG ekspor dengan dua task, sedangkan versi yang siap produksi membutuhkan dua penjaga tambahan. Penjaga pertama adalah FileSensor bernama tunggu_ingesta yang menunggu folder ingesta milik data interval dag run, sehingga DAG ekspor dapat dipicu kapan pun tanpa risiko membaca folder yang belum terbentuk. Penjaga kedua adalah task periksa_mutu yang memeriksa rentang nilai sebelum perhitungan, sehingga data cacat berhenti di tengah alur dan tidak pernah sampai ke berkas ekspor. Alur lengkapnya menjadi lima task, yaitu menunggu, membaca, memeriksa, merangkum, lalu mengekspor.

Jalur folder pada sensor ditulis sebagai template {{ ds_nodash }}, sehingga setiap dag run menunggu folder partisinya sendiri dan run untuk tanggal berbeda tidak saling mengganggu. Task ekspor menggunakan kebiasaan yang sama pada berkas hasilnya, yaitu folder dipartisi per tanggal dan ditulis dalam mode timpa. Cara ini menjadikan seluruh DAG idempoten sebagaimana disepakati pada rancangan.

python dag_ringkasan_ekspor.py 
"""DAG ringkasan dan ekspor proyek statistik ujian sekolah.

Menunggu folder ingesta tanggal data terbentuk, menahan alur lewat
gerbang mutu, lalu menyusun ringkasan lintas kelas dan mengekspornya
sebagai berkas CSV terpartisi.
"""
import csv
import os

import pendulum
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.sdk import dag, task, get_current_context

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


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_ringkasan_ekspor",
    schedule=None,
    start_date=pendulum.datetime(2026, 11, 1, tz="UTC"),
    description="Sensor ingesta, gerbang mutu, ringkasan lintas kelas, lalu ekspor CSV.",
)
def dag_ringkasan_ekspor():
    tunggu = FileSensor(
        task_id="tunggu_ingesta",
        filepath=f"{FOLDER_INGESTA}/{{{{ ds_nodash }}}}",
        poke_interval=1,
        timeout=15,
        mode="poke",
    )

    @task
    def baca_data() -> dict:
        """Membaca nilai dan kehadiran dari folder ingesta tanggal data."""
        context = get_current_context()
        folder = f"{FOLDER_INGESTA}/{context['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 ringkas(data: dict) -> dict:
        """Menyusun rata-rata nilai dan persen 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

    @task
    def ekspor(hasil: dict) -> str:
        """Menuliskan ringkasan ke berkas CSV terpartisi tanggal data."""
        context = get_current_context()
        folder = f"{FOLDER_EKSPOR}/{context['ds_nodash']}"
        os.makedirs(folder, exist_ok=True)
        path = f"{folder}/ringkasan.csv"
        with open(path, "w", newline="", encoding="utf-8") as berkas:
            penulis = csv.writer(berkas)
            penulis.writerow(["kelas", "rata_nilai", "persen_hadir"])
            for k in sorted(hasil):
                penulis.writerow([k, hasil[k]["rata_nilai"], hasil[k]["persen_hadir"]])
        print(f"ringkasan diekspor ke {path}")
        return path

    data = baca_data()
    tunggu >> data
    ekspor(ringkas(periksa_mutu(data)))


dag_ringkasan_ekspor()

Lingkungan latihan dan struktur DAG#

Lingkungan latihan topik ini disiapkan terpisah. Koneksi sistem berkas fs_default yang dibutuhkan FileSensor dideklarasikan melalui variabel lingkungan, folder dags disiapkan, dan basis data metadata dimigrasikan sekali. Data ingesta partisi 20261104 dituliskan langsung dengan isi yang sama seperti pada topik arsitektur, sehingga latihan ini tidak menuntut DAG ingesta berjalan lebih dulu.

bash 
# lingkungan latihan terpisah untuk topik ini
export AIRFLOW_HOME=/tmp/ra-implementasi
export AIRFLOW__CORE__LOAD_EXAMPLES=False
# koneksi sistem berkas fs_default dibaca dari variabel lingkungan
export AIRFLOW_CONN_FS_DEFAULT='fs:///'
bash 
# salin berkas DAG ke folder yang dipindai Airflow
export AIRFLOW_HOME=/tmp/ra-implementasi
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_ringkasan_ekspor.py "$AIRFLOW_HOME/dags/"
bash 
# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
export AIRFLOW_HOME=/tmp/ra-implementasi
airflow db migrate > /dev/null 2>&1
bash 
# isi folder ingesta partisi 20261104, meniru hasil DAG ingesta harian
mkdir -p /tmp/ra-statistik/ingesta/20261104
printf 'nis,nama,kelas,nilai\n1001,Ali,A,82\n1002,Budi,A,74\n1003,Citra,A,91\n1004,Dewi,B,66\n1005,Eka,B,78\n1006,Fani,B,85\n' > /tmp/ra-statistik/ingesta/20261104/nilai.csv
printf 'nis,tanggal,status\n1001,2026-11-04,hadir\n1002,2026-11-04,hadir\n1003,2026-11-04,izin\n1004,2026-11-04,hadir\n1005,2026-11-04,hadir\n1006,2026-11-04,sakit\n' > /tmp/ra-statistik/ingesta/20261104/kehadiran.csv

Struktur DAG yang sudah dimuat diperiksa tanpa mengeksekusi task apa pun. Lima task yang muncul memperlihatkan dua penjaga tambahan di samping tiga task pekerja.

bash 
# muat modul DAG lalu cetak dag_id, jadwal, dan seluruh task_id
export AIRFLOW_HOME=/tmp/ra-implementasi
python3 -c "import dag_ringkasan_ekspor; d = dag_ringkasan_ekspor.dag_ringkasan_ekspor(); print(d.dag_id); print(d.schedule); print(sorted(t.task_id for t in d.tasks))"
text Keluaran 
dag_ringkasan_ekspor
None
['baca_data', 'ekspor', 'periksa_mutu', 'ringkas', 'tunggu_ingesta']

Run pertama, data sah diekspor#

DAG dijalankan untuk pemicuan 2026-11-04. Sensor menemukan folder partisinya pada pemeriksaan pertama karena data ingesta sudah disiapkan, sehingga seluruh lima task berjalan tanpa menunggu.

bash 
# jalankan seluruh task DAG untuk pemicuan 2026-11-04
# log panjang dan bertanda waktu, sehingga diarahkan ke berkas
export AIRFLOW_HOME=/tmp/ra-implementasi
export AIRFLOW__CORE__LOAD_EXAMPLES=False
export AIRFLOW_CONN_FS_DEFAULT='fs:///'
airflow dags test dag_ringkasan_ekspor 2026-11-04 > /tmp/ra-implementasi/run-sah.log 2>&1

Bukti keberhasilannya dibaca dari log dan dari berkas hasil ekspor.

bash 
# hitung task yang berstatus success pada log
grep -c "new_state=success" /tmp/ra-implementasi/run-sah.log
# tampilkan berkas ringkasan yang diekspor
cat /tmp/ra-statistik/ekspor/20261104/ringkasan.csv
text Keluaran 
5
kelas,rata_nilai,persen_hadir
A,82.3,67
B,76.3,67

Angka 5 pada hitungan pertama menyatakan lima task berstatus success, termasuk sensor dan gerbang mutunya. Isi berkas ringkasan dapat dihitung ulang langsung dari data ingesta. Kelas A memuat nilai 82, 74, dan 91 yang rata-ratanya 82.3, dengan dua dari tiga siswa berstatus hadir sehingga persen kehadirannya 67. Kelas B memuat nilai 66, 78, dan 85 yang rata-ratanya 76.3 dengan pola kehadiran yang sama. Eksekusi ulang run ini aman karena seluruh task menimpa hasil partisi yang sama tanpa menambah isinya.

Run kedua, gerbang menolak data cacat#

Gerbang mutu diuji dengan data yang melanggar aturan, yaitu satu baris bernilai 150 yang berada di luar rentang yang disepakati. Data cacat dituliskan ke folder partisi 20261105, lalu DAG dijalankan untuk pemicuan tanggal itu.

bash 
# isi folder ingesta partisi 20261105 dengan satu baris cacat
mkdir -p /tmp/ra-statistik/ingesta/20261105
printf 'nis,nama,kelas,nilai\n1001,Ali,A,82\n1004,Dewi,B,150\n' > /tmp/ra-statistik/ingesta/20261105/nilai.csv
printf 'nis,tanggal,status\n1001,2026-11-05,hadir\n1004,2026-11-05,hadir\n' > /tmp/ra-statistik/ingesta/20261105/kehadiran.csv
bash 
# jalankan DAG untuk pemicuan 2026-11-05 yang berisi data cacat
# perintah diakhiri echo karena gerbang mutu membuatnya berakhir gagal
export AIRFLOW_HOME=/tmp/ra-implementasi
export AIRFLOW__CORE__LOAD_EXAMPLES=False
export AIRFLOW_CONN_FS_DEFAULT='fs:///'
airflow dags test dag_ringkasan_ekspor 2026-11-05 > /tmp/ra-implementasi/run-cacat.log 2>&1 || echo "dags test berakhir dengan task gagal"
text Keluaran 
dags test berakhir dengan task gagal

Sensor kali ini juga berhasil karena folder partisinya ada, tetapi alur berhenti di gerbang. Pesan penolakannya terbaca pada log bersama jumlah task yang gagal, dan folder ekspor tidak berubah.

bash 
# tampilkan pesan penolakan gerbang dan hitung task yang gagal
grep "RuntimeError: gerbang" /tmp/ra-implementasi/run-cacat.log | head -n 1
grep -c "new_state=failed" /tmp/ra-implementasi/run-cacat.log
# tampilkan isi folder ekspor
ls /tmp/ra-statistik/ekspor/
text Keluaran 
RuntimeError: gerbang mutu gagal: 1 baris tidak memenuhi aturan
1
20261104

Ketiga bukti tersebut saling melengkapi. Satu task berstatus failed, yaitu periksa_mutu, sedangkan task ringkas dan ekspor tidak pernah berjalan karena task hulu keduanya gagal. Folder ekspor hanya memuat partisi run yang sah. Data cacat berhenti di tengah pipeline tanpa menyentuh satu pun hasil, dan perbaikannya cukup dilakukan dengan memperbaiki data sumber lalu menjalankan ulang run tersebut.

Kesalahan umum#

  • Sensor yang menunggu jalur tanpa template. FileSensor dengan filepath tetap memeriksa folder yang sama pada setiap run, sehingga run tanggal berikutnya berhasil melewati sensor meskipun data tanggalnya belum ada. Gunakan template konteks seperti {{ ds_nodash }} agar setiap run menunggu partisinya sendiri.
  • Gerbang mutu ditempatkan setelah pekerjaan berat. Penolakan yang terlambat membuat seluruh perhitungan ikut berjalan untuk data yang akhirnya dibuang. Letakkan gerbang tepat setelah data dibaca dan sebelum task transformasi pertama.
  • Ekspor yang menambah baris pada berkas bersama. Run ulang untuk tanggal yang sama melipatgandakan isi berkas tanpa pesan error apa pun. Tulis berkas ekspor dengan mode timpa ke folder terpartisi tanggal, sebagaimana task ekspor pada DAG ini.

Ringkasan#

  • DAG dag_ringkasan_ekspor terdiri atas lima task, yaitu FileSensor yang menunggu folder ingesta partisi tanggal, pembacaan data, gerbang mutu, perangkuman per kelas, dan ekspor CSV terpartisi.
  • Run dengan data sah mengekspor ringkasan.csv berisi rata-rata nilai dan persen kehadiran per kelas, dan hasilnya dapat dihitung ulang langsung dari data sumber.
  • Run dengan satu baris bernilai 150 ditolak oleh gerbang melalui RuntimeError, task hilir tidak berjalan, dan folder ekspor tidak berubah.
  • Seluruh task menulis hasil dalam mode timpa ke jalur yang memuat ds_nodash, sehingga run ulang untuk tanggal yang sama menghasilkan keadaan akhir yang sama.

Referensi#