Gudang data tanpa pemeriksaan hanya menunda kejutan: baris hilang senyap pada Bagian 4 dan versi dimensi yang salah pada Bagian 6 sama-sama mungkin terulang tanpa pesan galat apa pun. Karena tabel kita sengaja tanpa constraint, tanggung jawab integritas sepenuhnya berpindah ke pemeriksaan kualitas data yang berjalan setiap akhir pipeline. Pada bagian ini kita menulis lima pemeriksaan pada src/quality.py, memastikan pelanggaran benar-benar terdeteksi lewat pytest, dan menghasilkan laporan JSON pada data/dq_reports/. Kerangka berpikirnya mengikuti chapter Data Validation dan Data Consistency, sementara padanan dbt-nya adalah data tests yang dibahas chapter Testing Model Data.

Lima Pemeriksaan #

PemeriksaanPertanyaan yang dijawab
rekonsiliasi_barisapakah jumlah baris fact sama dengan jumlah baris sumber staging?
orphan_foreign_keyapakah setiap foreign key fact menunjuk baris dimensi yang ada?
measure_tidak_nullapakah kolom measure dan kunci fact bebas NULL?
surrogate_key_unikapakah surrogate key tiap dimensi unik dan grain fact tidak berganda?
status_validapakah kolom status hanya memuat tiga nilai resmi?

Kelima pemeriksaan berbentuk fungsi Python yang mengembalikan kamus nama, lulus, dan detail. Sebagai contoh, pemeriksaan orphan foreign key memeriksa keempat relasi sekaligus:

python 
def cek_orphan_fk(con) -> dict:
    """Setiap foreign key di fact harus menunjuk baris dimensi yang ada."""
    per_fk = {}
    for kolom_fk, tabel_dim, kolom_sk in FK_FACT:
        yatim = con.execute(
            f"""
            SELECT COUNT(*) FROM fact_pesanan AS f
            LEFT JOIN {tabel_dim} AS d ON d.{kolom_sk} = f.{kolom_fk}
            WHERE d.{kolom_sk} IS NULL
            """
        ).fetchone()[0]
        per_fk[kolom_fk] = int(yatim)
    return {
        "nama": "orphan_foreign_key",
        "lulus": sum(per_fk.values()) == 0,
        "detail": ", ".join(f"{k}={v}" for k, v in per_fk.items()),
    }

Mari kita telaah mekanismenya:

  1. Baris for kolom_fk, tabel_dim, kolom_sk in FK_FACT: daftar pemetaan empat foreign key didefinisikan sebagai data, sehingga menambah relasi baru cukup menambah satu entri daftar.
  2. Baris LEFT JOIN ... WHERE d.{kolom_sk} IS NULL: pola anti-join; baris fact yang pasangan dimensinya tidak ada akan tetap hidup di sisi kiri dengan kolom dimensi NULL.
  3. Baris "lulus": sum(per_fk.values()) == 0: pemeriksaan lulus hanya bila tidak ada satu pun orphan pada keempat relasi.
  4. Baris "detail": ", ".join(...): rincian per foreign key ikut dilaporkan agar penelusuran langsung menuju relasi yang bermasalah.

Laporan Kualitas sebagai JSON #

jalankan_pemeriksaan menjalankan kelima fungsi lalu menulis laporan melalui simpan_laporan. Berikut isi data/dq_reports/dq_report.json hasil pipeline Bagian 5:

json 
{
  "diperiksa_pada": "2026-10-17T07:56:03+00:00",
  "semua_lulus": true,
  "pemeriksaan": [
    {
      "nama": "rekonsiliasi_baris",
      "lulus": true,
      "detail": "fact 32 baris vs sumber stg_pesanan 32 baris"
    },
    {
      "nama": "orphan_foreign_key",
      "lulus": true,
      "detail": "tanggal_sk=0, pelanggan_sk=0, produk_sk=0, toko_sk=0"
    },
    {
      "nama": "measure_tidak_null",
      "lulus": true,
      "detail": "0 baris fact memuat kolom measure yang NULL"
    },
    {
      "nama": "surrogate_key_unik",
      "lulus": true,
      "detail": "dim_tanggal duplikat=0, dim_pelanggan duplikat=0, dim_produk duplikat=0, dim_toko duplikat=0, fact_pesanan duplikat=0"
    },
    {
      "nama": "status_valid",
      "lulus": true,
      "detail": "0 baris fact memuat status di luar ['selesai', 'dibatalkan', 'dikembalikan']"
    }
  ]
}

Berkas di atas adalah isi utuh data/dq_reports/dq_report.json; kolom diperiksa_pada mencatat waktu pemeriksaan sehingga laporan lama dan baru mudah dibandingkan. Berkas JSON ini adalah artefak yang bisa dibaca pipeline lain atau sistem peringatan; kolom semua_lulus menjadi satu nilai yang cukup diperiksa untuk memutuskan pipeline lanjut atau berhenti.

Kualitas Data sebagai Gerbang Pipeline #

Pemeriksaan ini bukan skrip lepasan yang dijalankan kadang-kadang; src/pipeline.py memanggilnya sebagai tahap kelima dan terakhir setiap kali gudang dibangun ulang:

python 
        laporan = (Path(berkas_laporan) if berkas_laporan
                   else quality.DIREKTORI_LAPORAN / "dq_report.json")
        hasil_kualitas = quality.jalankan_pemeriksaan(con, simpan_ke=laporan)

Parameter berkas_laporan membuat uji end-to-end dapat mengarahkan laporan ke berkas sementara tanpa menyentuh laporan proyek. Hasilnya tercetak pada keluaran pipeline Bagian 5 sebagai baris [5/5] kualitas data: 5/5 pemeriksaan lulus. Karena jalankan_pemeriksaan mengembalikan daftar hasil, pemanggil bebas menentukan kebijakan: pipeline latihan kita hanya mencatat, sedangkan pipeline produksi lazim menghentikan langkah berikutnya ketika quality.semua_lulus(hasil) bernilai False. Pemisahan antara “mengukur” pada quality.py dan “memutuskan” pada pemanggil inilah yang membuat kebijakan berubah tanpa menyentuh pemeriksaan sedikit pun.

Menguji Bahwa Pelanggaran Terdeteksi #

Pemeriksaan yang selalu lulus belum terbukti bekerja, jadi tests/test_quality.py sengaja menyuntikkan data rusak lalu memastikan pemeriksaan menolaknya:

python 
def test_orphan_fk_terdeteksi(con):
    _siapkan_gudang(con)
    con.execute(
        "INSERT INTO fact_pesanan VALUES "
        "('ORD-7003', 1, 20250210, 999, 1, 1, 1, 85000, 0, 85000, 'selesai')"
    )
    hasil = quality.cek_orphan_fk(con)
    assert hasil["lulus"] is False
    assert "pelanggan_sk=1" in hasil["detail"]

Baris yang disisipkan memakai pelanggan_sk = 999 yang tidak ada di dimensi mana pun; tes lulus hanya bila pemeriksaan gagal dengan rincian yang benar. Pola yang sama dipakai untuk measure NULL, surrogate key ganda, baris fact yang hilang, dan status di luar daftar resmi. Seluruh rangkaian uji proyek kemudian dijalankan:

bash 
python3 -m pytest tests/ -q

Output yang Dihasilkan:

text 
....................................                                     [100%]
36 passed in 2.81s

Tiga puluh enam uji mencakup dimensi, fact table, SCD Type 2, kualitas data, dan pipeline end-to-end. Ringkasan per berkas: test_dim.py menguji surrogate key dan idempotensi, test_fact.py menguji resolusi foreign key dan grain, test_scd2.py menguji penutupan versi dan pemilihan versi sesuai tanggal, test_quality.py menguji deteksi pelanggaran, dan test_pipeline_e2e.py menjalankan pipeline utuh dua kali untuk membuktikan idempotensi. Dokumentasi pytest ada di https://docs.pytest.org/ , dan padanan dbt test dengan data tests generik di https://docs.getdbt.com/docs/build/data-tests .

Gudang data kini teruji dan terdokumentasi kualitasnya. Pertanyaan terakhir sebelum produksi: apakah query analitik di atasnya berjalan cukup cepat? Bagian 8 mengukurnya dengan EXPLAIN ANALYZE.