Transformasi pada chapter sebelumnya menghasilkan data yang tampak bersih, tetapi “tampak” bukan kata yang cukup untuk data yang akan diekspor ke tim lain. Satu kolom yang tidak terbentuk, satu nilai jumlah berisi 0, atau satu id_pesanan ganda cukup untuk membuat laporan omzet memberi angka yang salah — dan kesalahan seperti ini biasanya baru ketahuan berhari-hari kemudian. Solusinya adalah memasang gerbang validasi: pemeriksaan otomatis yang menilai setiap DataFrame sebelum diteruskan ke tahap ekspor, dan menolak data yang tidak memenuhi syarat dengan pesan yang jelas. Chapter ini membangun dua lapis gerbang, lapis schema dan lapis aturan bisnis, lengkap dengan contoh data yang lolos dan data yang ditolak.

Dua Lapis Gerbang #

Kedua lapis memeriksa hal yang berbeda dan dijalankan dengan urutan tetap:

LapisYang diperiksaContoh temuan
SchemaKelengkapan nama kolomKolom produk tidak ada
Aturan bisnisIsi data terhadap aturan tokojumlah <= 0, id_pesanan duplikat, cabang di luar daftar resmi

Urutan ini penting: pemeriksaan aturan bisnis menulis df["jumlah"], dan penulisan itu melempar KeyError bila kolom jumlah belum ada. Schema gate dijalankan lebih dulu sehingga pipeline berhenti dengan pesan yang menjelaskan kolom yang hilang, bukan KeyError yang menuntut pembacaan traceback.

Implementasi Gerbang Validasi #

Script berikut mendefinisikan validasi_schema() yang mengembalikan daftar kolom hilang, validasi_aturan() yang mengumpulkan pelanggaran, serta gerbang() yang melempar ValueError bila ada temuan. Ketiganya lalu diuji terhadap satu DataFrame sehat dan dua DataFrame rusak.

python 
"""Gerbang validasi schema dan aturan bisnis untuk hasil pipeline."""
import pandas as pd

KOLOM_WAJIB = ["id_pesanan", "cabang", "produk", "jumlah", "harga_satuan"]
CABANG_RESMI = {"Jakarta", "Bandung", "Surabaya"}


def pesanan_bersih() -> pd.DataFrame:
    """Hasil transform yang sehat dari chapter sebelumnya."""
    return pd.DataFrame({
        "id_pesanan": ["P001", "P002", "P003", "P004", "P005", "P006"],
        "cabang": ["Jakarta", "Bandung", "Surabaya", "Jakarta", "Bandung", "Surabaya"],
        "produk": ["Kopi Arabika", "Teh Melati", "Beras Premium",
                   "Gula Pasir", "Kopi Arabika", "Minyak Goreng"],
        "jumlah": [2, 3, 1, 4, 2, 3],
        "harga_satuan": [45000, 18000, 62000, 14000, 45000, 32000],
        "total": [90000, 54000, 62000, 56000, 90000, 96000],
    })


def validasi_schema(df: pd.DataFrame, kolom_wajib: list[str] = KOLOM_WAJIB) -> list[str]:
    """Kembalikan daftar kolom wajib yang tidak ada di DataFrame."""
    return [kolom for kolom in kolom_wajib if kolom not in df.columns]


def validasi_aturan(df: pd.DataFrame) -> list[str]:
    """Periksa aturan bisnis; kembalikan daftar pelanggaran."""
    masalah: list[str] = []
    buruk = df.loc[df["jumlah"] <= 0, "id_pesanan"].tolist()
    if buruk:
        masalah.append(f"jumlah <= 0 pada {buruk}")
    duplikat = df.loc[df["id_pesanan"].duplicated(), "id_pesanan"].unique().tolist()
    if duplikat:
        masalah.append(f"id_pesanan duplikat: {duplikat}")
    asing = df.loc[~df["cabang"].isin(CABANG_RESMI), "cabang"].unique().tolist()
    if asing:
        masalah.append(f"cabang tidak dikenal: {asing}")
    return masalah


def gerbang(df: pd.DataFrame) -> None:
    """Loloskan DataFrame hanya bila schema lalu aturan bersih."""
    hilang = validasi_schema(df)
    if hilang:
        raise ValueError(f"Validasi schema gagal: kolom wajib tidak ada: {hilang}")
    pelanggaran = validasi_aturan(df)
    if pelanggaran:
        raise ValueError("Validasi aturan gagal: " + "; ".join(pelanggaran))


if __name__ == "__main__":
    bersih = pesanan_bersih()
    print("== Data bersih ==")
    print(f"schema : LOLOS ({len(validasi_schema(bersih))} kolom hilang)")
    print(f"aturan : LOLOS ({len(validasi_aturan(bersih))} pelanggaran)")
    gerbang(bersih)
    print(f"gerbang: {len(bersih)} baris dinyatakan siap diekspor")
    print()

    rusak_kolom = pesanan_bersih().drop(columns="produk")
    print("== Data rusak 1 (kolom hilang) ==")
    print(f"schema : {validasi_schema(rusak_kolom)}")
    try:
        gerbang(rusak_kolom)
    except ValueError as e:
        print(f"ValueError: {e}")
    print()

    rusak_isi = pesanan_bersih().copy()
    rusak_isi.loc[2, "jumlah"] = 0
    rusak_isi = pd.concat([rusak_isi, rusak_isi.iloc[[1]]], ignore_index=True)
    rusak_isi.loc[5, "cabang"] = "Medan"
    print("== Data rusak 2 (isi tidak wajar) ==")
    print(f"aturan : {validasi_aturan(rusak_isi)}")
    try:
        gerbang(rusak_isi)
    except ValueError as e:
        print(f"ValueError: {e}")

Output yang Dihasilkan:

text 
== Data bersih ==
schema : LOLOS (0 kolom hilang)
aturan : LOLOS (0 pelanggaran)
gerbang: 6 baris dinyatakan siap diekspor

== Data rusak 1 (kolom hilang) ==
schema : ['produk']
ValueError: Validasi schema gagal: kolom wajib tidak ada: ['produk']

== Data rusak 2 (isi tidak wajar) ==
aturan : ["jumlah <= 0 pada ['P003']", "id_pesanan duplikat: ['P002']", "cabang tidak dikenal: ['Medan']"]
ValueError: Validasi aturan gagal: jumlah <= 0 pada ['P003']; id_pesanan duplikat: ['P002']; cabang tidak dikenal: ['Medan']

Bagian yang layak dibedah:

  1. validasi_schema() hanya membandingkan dua daftar: kolom_wajib dan df.columns. Fungsi ini mengembalikan daftar temuan alih-alih langsung melempar error, sehingga pemanggil bebas memutuskan apa yang dilakukan dengan temuan tersebut.
  2. validasi_aturan() memakai boolean masking untuk setiap aturan: df["jumlah"] <= 0, df["id_pesanan"].duplicated(), dan ~df["cabang"].isin(CABANG_RESMI). Hasil filter dikonversi menjadi daftar id atau nama cabang, sehingga pesan pelanggaran langsung menunjuk baris yang bermasalah.
  3. gerbang() menggabungkan keduanya: schema dulu, lalu aturan, dan setiap kegagalan melempar ValueError dengan pesan yang menyebut semua temuan sekaligus.

Pada DataFrame rusak pertama, kolom produk dihapus dengan drop(columns="produk"); gerbang langsung berhenti di lapis schema dengan pesan Validasi schema gagal: kolom wajib tidak ada: ['produk']. Pada DataFrame rusak kedua, schema lengkap tetapi isinya bermasalah; pesan error menjelaskan tiga pelanggaran sekaligus, termasuk baris P003 yang jumlah-nya 0 dan cabang Medan yang tidak terdaftar.

Merancang Aturan Bisnis #

Aturan yang tepat berasal dari kontrak antara pipeline dan pemakai laporannya. Untuk data pesanan Toko Berkah Abadi, kontraknya bisa dirumuskan sebagai berikut:

  • Keutuhan: setiap id_pesanan unik, karena id adalah kunci yang dirujuk sistem lain.
  • Kewajaran nilai: jumlah lebih besar dari nol; harga dan total tidak boleh negatif.
  • Kesatuan domain: cabang hanya boleh bernilai dari daftar resmi, sehingga salah ketik seperti Medan tertangkap sebelum masuk laporan.

Prinsip penerapannya: setiap aturan menghasilkan satu pesan yang menyebut data pelanggar, dan gerbang mengumpulkan semua pelanggaran sebelum menolak data. Developer yang menerima pesan jumlah <= 0 pada ['P003'] tahu persis baris mana yang harus diperbaiki, tanpa perlu membuka data secara manual.

Gerbang yang menolak data perlu tahu cara memberi tahu manusia apa yang terjadi — mencatat perjalanan pipeline dan menangkap error tahap demi tahap. Itu pekerjaan module logging dan error handling, topik chapter berikutnya.

Referensi resmi: pandas.DataFrame.isin , pandas.DataFrame.duplicated , dan user guide Gotchas .