KAL
Menu

Gerbang mutu data

Menempatkan pemeriksaan mutu data yang menghentikan alur sebelum data buruk menyebar ke hilir.

Daftar isi

Task yang menghasilkan data untuk sistem lain tidak cukup berstatus success, karena pernyataan SQL yang benar sintaksnya tetap dapat menghasilkan data yang salah isinya, misalnya nilai kosong atau angka di luar rentang wajar. Topik ini membahas gerbang mutu data, yaitu pemeriksaan yang ditempatkan sebagai task eksplisit di antara pemuatan data dan task hilirnya. Pemeriksaan tersebut mengikuti prinsip fail-fast, yakni menghentikan eksekusi secepatnya pada titik pelanggaran pertama. Pembahasannya mencakup pemeriksaan manual dengan TaskFlow, perhitungan jumlah pelanggaran melalui SQL, operator pemeriksaan terstruktur dari provider common.sql, serta penempatan gerbang pada graf DAG. Setelah gerbang ini diterapkan, data buruk berhenti di gerbang dan task hilir tidak ikut memprosesnya.

Prinsip fail-fast pada alur data#

Gerbang mutu bekerja dengan mengubah aturan data menjadi kondisi keberhasilan task. Aturannya sederhana, yaitu jumlah baris yang melanggar harus nol, dan task pemeriksaan melemparkan error apabila jumlahnya lebih dari nol. Karena Airflow hanya menjalankan task hilir setelah seluruh task upstream langsungnya berstatus success, satu kegagalan pada gerbang otomatis menghentikan seluruh rantai task di hilir tanpa logika tambahan.

Diagram berikut menempatkan gerbang tepat setelah task pemuatan dan sebelum semua konsumen data. Pelanggaran menghentikan dag run pada gerbang dengan status failed, sedangkan data yang sehat mengalir ke hilir seperti biasa.

muat_dataperiksa_mutubuat_ringkasansehatfailed
Posisi gerbang mutu pada graf DAG, dengan task hilir yang hanya berjalan setelah pemeriksaan lolos

Bentuk task pemeriksa bebas, mulai dari fungsi @task yang memeriksa data di memori hingga query SQL yang menghitung pelanggaran langsung di basis data. Pilihan mengikuti letak datanya, yaitu pemeriksaan di memori untuk hasil task sebelumnya dan pemeriksaan SQL untuk data yang sudah berada di dalam basis data.

Gerbang dengan task pemeriksaan#

Berkas dag_gerbang.py berikut memuat tiga task, yaitu menyiapkan data nilai ujian, memeriksa mutu datanya, lalu membuat ringkasan. Task muat_nilai menggantikan tahap pemuatan sungguhan agar latihan tetap berjalan mandiri.

python dag_gerbang.py 
from airflow.sdk import dag, task


@dag(
    dag_id="dag_gerbang",
    schedule=None,
    description="DAG latihan: gerbang mutu sebelum task hilir.",
)
def dag_gerbang():
    @task
    def muat_nilai() -> list[dict]:
        """Menyiapkan data yang akan diperiksa."""
        return [
            {"murid": "Ali", "nilai": 82},
            {"murid": "Budi", "nilai": 74},
            {"murid": "Citra", "nilai": 91},
            {"murid": "Dewi", "nilai": 66},
            {"murid": "Eka", "nilai": 78},
        ]

    @task
    def periksa_mutu(baris: list[dict]) -> int:
        """Menghentikan alur bila ada nilai hilang atau di luar rentang."""
        pelanggaran = [
            b["murid"]
            for b in baris
            if b["nilai"] is None or not 0 <= b["nilai"] <= 100
        ]
        print(f"Jumlah pelanggaran: {len(pelanggaran)}")
        if pelanggaran:
            raise RuntimeError(f"gerbang mutu gagal: {len(pelanggaran)} baris tidak memenuhi aturan")
        return len(baris)

    @task
    def buat_ringkasan(baris: list[dict]) -> None:
        """Task hilir yang hanya berjalan bila gerbang mutu lolos."""
        rata_rata = round(sum(b["nilai"] for b in baris) / len(baris), 1)
        print(f"Rata-rata kelas: {rata_rata}")

    baris = muat_nilai()
    gerbang = periksa_mutu(baris)
    gerbang >> buat_ringkasan(baris)


dag_gerbang()

Aturan mutu dinyatakan dalam satu kondisi, yaitu setiap nilai harus terisi dan berada di rentang 0 sampai 100. Task periksa_mutu mengumpulkan nama murid yang melanggar, mencetak jumlahnya ke log, lalu melemparkan RuntimeError apabila daftarnya tidak kosong. Baris terakhir gerbang >> buat_ringkasan(baris) membuat task ringkasan bergantung pada gerbang, sehingga kegagalan gerbang ikut menghentikan ringkasan.

bash 
# lingkungan latihan terpisah agar tidak bercampur dengan instalasi lain
export AIRFLOW_HOME=/tmp/ra-gerbang-mutu-data
export AIRFLOW__CORE__LOAD_EXAMPLES=False
bash 
# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
airflow db migrate > /dev/null 2>&1
bash 
# uji bahwa berkas DAG dapat diparse tanpa keluaran apa pun
python3 dag_gerbang.py
bash 
# salin berkas DAG ke folder yang dipindai Airflow
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_gerbang.py "$AIRFLOW_HOME/dags/"
bash 
# jalankan seluruh task DAG sekali secara lokal
# log panjang dan bertanda waktu, sehingga diarahkan ke berkas
airflow dags test dag_gerbang 2026-11-04 > "$AIRFLOW_HOME/dags-test.log" 2>&1
grep -oE "Jumlah pelanggaran: [0-9]+|Rata-rata kelas: [0-9.]+" "$AIRFLOW_HOME/dags-test.log"
text Keluaran 
Jumlah pelanggaran: 0
Rata-rata kelas: 78.2

Log dari perintah tersebut panjang dan bertanda waktu, sehingga diarahkan ke berkas dan hanya dua baris kunci yang ditampilkan. Pada log tersebut, task periksa_mutu mencetak jumlah pelanggaran, sedangkan task buat_ringkasan mencetak rata-rata kelas. Ketiga task berstatus success, dan dag run ditutup dengan status yang sama.

Perilaku sebaliknya terjadi ketika salah satu nilai diganti menjadi 166 atau None pada data muat_nilai. Setelah DAG diuji ulang, task periksa_mutu berhenti dengan RuntimeError: gerbang mutu gagal: 1 baris tidak memenuhi aturan. Status task tersebut failed, task buat_ringkasan tidak dijalankan sama sekali, dan dag run berakhir failed. Inilah bentuk paling sederhana dari fail-fast, yaitu error yang muncul di titik masalahnya, bukan sebagai angka janggal pada laporan di hilir.

Menghitung pelanggaran dengan SQL#

Apabila data sudah berada di dalam basis data, pemeriksaan yang sama ditulis sebagai query yang mengembalikan jumlah pelanggaran. Blok berikut memeriksa aturan rentang dan keberadaan nilai pada tabel yang datanya memenuhi aturan, dan angka nol berarti gerbang lolos.

sql 
CREATE TABLE nilai_ujian (murid TEXT, kelas TEXT, nilai INTEGER);
INSERT INTO nilai_ujian VALUES
  ('Ali', 'A', 82), ('Budi', 'A', 74), ('Citra', 'B', 91),
  ('Dewi', 'B', 66), ('Eka', 'A', 78);
SELECT COUNT(*) AS pelanggaran
FROM nilai_ujian
WHERE nilai IS NULL OR nilai < 0 OR nilai > 100;
text Keluaran 
0

Blok kedua memuat satu baris yang melanggar, yaitu Eka dengan nilai 105. Query yang sama menampilkan baris tersebut sebagai bukti deteksi sebelum task hilir dipicu.

sql 
CREATE TABLE nilai_ujian (murid TEXT, kelas TEXT, nilai INTEGER);
INSERT INTO nilai_ujian VALUES
  ('Ali', 'A', 82), ('Budi', 'A', 74), ('Citra', 'B', 91),
  ('Dewi', 'B', 66), ('Eka', 'A', 105);
SELECT murid, nilai
FROM nilai_ujian
WHERE nilai IS NULL OR nilai < 0 OR nilai > 100;
text Keluaran 
Eka|105

Dalam DAG, query semacam ini dijalankan oleh SQLExecuteQueryOperator dengan handler=fetch_all_handler, lalu task pemeriksa membaca hasilnya dan melemparkan error apabila jumlahnya lebih dari nol. Bentuk deteksi kedua yang menampilkan baris pelanggar berguna untuk pesan error karena menyebut data yang bermasalah, bukan sekadar jumlahnya.

Operator pemeriksaan terstruktur#

Provider common.sql menyediakan dua operator yang membangun pemeriksaan semacam itu dari kamus Python, sehingga aturan mutu terbaca seperti daftar pernyataan. SQLColumnCheckOperator memeriksa kolom tertentu melalui column_mapping, sedangkan SQLTableCheckOperator memeriksa keseluruhan tabel melalui checks berupa pernyataan SQL yang harus bernilai benar. Keduanya melemparkan error secara otomatis apabila ada pemeriksaan yang gagal, dan kamus aturannya menerima beberapa aturan sekaligus dalam satu task.

python Contoh: pemeriksaan kolom dan tabel 
from airflow.providers.common.sql.operators.sql import SQLColumnCheckOperator, SQLTableCheckOperator

periksa_kolom = SQLColumnCheckOperator(
    task_id="periksa_kolom",
    conn_id="nilai_sqlite",
    table="nilai_ujian",
    column_mapping={
        "nilai": {
            "null_check": {"equal_to": 0},
            "min": {"geq_to": 0},
            "max": {"leq_to": 100},
        }
    },
)

periksa_tabel = SQLTableCheckOperator(
    task_id="periksa_tabel",
    conn_id="nilai_sqlite",
    table="nilai_ujian",
    checks={"ada_baris": {"check_statement": "COUNT(*) >= 1"}},
)

Pada column_mapping, kunci luar adalah nama kolom, sedangkan kunci dalam adalah jenis pemeriksaannya, misalnya null_check yang mengharuskan jumlah NULL sama dengan nol, atau min dan max yang menjaga rentang nilai. Bentuk kamus ini menyerupai definisi test pada proyek dbt, sehingga aturan mutu yang sudah ditulis di proyek dbt mudah diterjemahkan menjadi task Airflow untuk tabel yang tidak dikelola oleh dbt.

Posisi gerbang pada graf#

Gerbang memberi jaminan hanya apabila posisinya benar pada graf. Aturannya, gerbang berada di hilir langsung sumber data yang diperiksa dan di hulu seluruh task yang menggunakan data tersebut, baik melalui rantai tunggal maupun melalui dependensi data TaskFlow. Apabila ada task hilir yang tidak bergantung pada gerbang, misalnya berjalan paralel dengan pemeriksaan tanpa menunggu gerbang, data buruk dapat mencapai task itu sebelum pemeriksaan sempat dijalankan.

Untuk tabel yang sama, satu gerbang cukup di awal rantai, karena semua task setelahnya mewarisi jaminan tersebut. Menambah pemeriksaan identik di tengah rantai hanya memperpanjang graf tanpa jaminan baru, kecuali data berubah lagi setelah gerbang.

Kesalahan umum#

  • Menempatkan gerbang di cabang terpisah yang tidak terhubung ke task hilir. Graf memuat task pemeriksa, tetapi task ringkasan tetap berjalan meskipun pemeriksaan gagal karena task tersebut tidak pernah menunggu gerbang. Perbaikannya dengan menjadikan seluruh konsumen data sebagai hilir langsung gerbang.
  • Memeriksa data setelah task hilir mengonsumsinya. Ringkasan sudah dibuat dari data yang belum diperiksa, sehingga hasilnya perlu dibuat ulang saat gerbang gagal. Perbaikannya dengan menempatkan pemeriksaan tepat setelah pemuatan dan sebelum semua konsumen.
  • Hanya menguji varian data yang sehat. Task pemeriksa yang selalu lolos belum membuktikan bahwa jalur errornya berfungsi. Perbaikannya dengan menguji varian yang melanggar secara terpisah, sebagaimana kedua kemungkinan pada contoh topik ini.

Ringkasan#

  • Gerbang mutu adalah task eksplisit yang melemparkan error saat aturan data dilanggar, dan kegagalannya menghentikan seluruh task hilir melalui dependensi biasa.
  • Pemeriksaan dinyatakan sebagai kondisi nol pelanggaran, baik pada data di memori melalui @task maupun pada basis data melalui query jumlah pelanggaran.
  • SQLColumnCheckOperator dan SQLTableCheckOperator dari provider common.sql membentuk pemeriksaan kolom dan tabel dari kamus aturan, dengan error otomatis saat ada pemeriksaan gagal.
  • Gerbang ditempatkan tepat setelah sumber data dan sebelum seluruh konsumen agar jaminannya berlaku, sedangkan pemeriksaan tanpa error hanya berupa catatan.

Referensi#