Data dari sumber eksternal berubah tanpa memberi tahu kita: kolom diganti nama, kolom angka berubah teks, baris muncul dua kali. Jika perubahan itu baru ketahuan di laporan akhir, menelusuri tahap mana yang membiarkannya lewat memakan waktu. Kontrak schema menutup celah ini: kita menuliskan bentuk data yang disepakati — kolom wajib, dtype, aturan isi — lalu memeriksanya di awal pipeline. Chapter ini membangun validator schema manual dengan alat yang sudah kita kenal dari chapter-chapter sebelumnya, lengkap dengan gerbang yang menolak data melanggar sebelum diproses lebih jauh.

Apa yang Masuk Kontrak Schema #

Kontrak schema yang berguna mencakup empat hal. Pertama, kolom wajib: daftar nama kolom yang harus ada agar transformasi berikutnya tidak menyandung KeyError. Kedua, dtype per kolom: janji bahwa jumlah benar-benar integer, bukan float atau teks. Ketiga, aturan isi: nilai tidak boleh negatif untuk kolom tertentu, dan kunci seperti kode_transaksi harus unik. Keempat, bentuk minimum: DataFrame tidak boleh kosong jika pipeline memang tidak mendukungnya. Semua aturan ini diperiksa dengan API yang sudah kita gunakan sebelumnya: df.columns, df.dtypes, Series.is_unique, dan mask boolean.

Validator dan Gerbang dalam Satu Skrip #

Skrip berikut mendefinisikan cek_schema() yang mengembalikan daftar pelanggaran, dan gerbang_schema() yang melempar ValueError bila daftarnya tidak kosong. Kontraknya disimpan sebagai dict Python biasa bernama SCHEMA_TRANSAKSI, sehingga perubahan kebutuhan hanya menyentuh satu tempat.

python 
import pandas as pd

SCHEMA_TRANSAKSI = {
    "kolom_wajib": ["kode_transaksi", "cabang", "jumlah", "harga"],
    "dtypes": {
        "kode_transaksi": "str",
        "cabang": "str",
        "jumlah": "Int64",
        "harga": "Int64",
    },
}


def cek_schema(df: pd.DataFrame, schema: dict) -> list[str]:
    """Kembalikan daftar pelanggaran kontrak; list kosong berarti valid."""
    pelanggaran: list[str] = []

    if df.empty:
        pelanggaran.append("DataFrame kosong")

    hilang = [k for k in schema["kolom_wajib"] if k not in df.columns]
    if hilang:
        pelanggaran.append(f"kolom hilang: {hilang}")

    for kolom, dtype in schema["dtypes"].items():
        if kolom in df.columns and df[kolom].dtype != dtype:
            pelanggaran.append(f"dtype {kolom}: {df[kolom].dtype} (harus {dtype})")

    for kolom in ("jumlah", "harga"):
        if kolom in df.columns and (df[kolom] < 0).any():
            pelanggaran.append(f"{kolom} memuat nilai negatif")

    if "kode_transaksi" in df.columns and not df["kode_transaksi"].is_unique:
        duplikat = df.loc[df["kode_transaksi"].duplicated(), "kode_transaksi"].tolist()
        pelanggaran.append(f"kode_transaksi duplikat: {duplikat}")

    return pelanggaran


def gerbang_schema(df: pd.DataFrame, schema: dict) -> pd.DataFrame:
    """Loloskan data yang valid; tolak sisanya dengan ValueError."""
    pelanggaran = cek_schema(df, schema)
    if pelanggaran:
        raise ValueError("schema tidak sesuai kontrak: " + "; ".join(pelanggaran))
    return df


transaksi_baik = pd.DataFrame(
    {
        "kode_transaksi": ["TRX-001", "TRX-002", "TRX-003"],
        "cabang": ["Jakarta", "Bandung", "Surabaya"],
        "jumlah": pd.array([2, 1, 4], dtype="Int64"),
        "harga": pd.array([10_000, 20_000, 2_500], dtype="Int64"),
    }
)
transaksi_buruk = pd.DataFrame(
    {
        "kode_transaksi": ["TRX-001", "TRX-001", None],
        "cabang": ["Jakarta", "Bandung", None],
        "jumlah": [2.5, -1.0, 4.0],
        "harga_satuan": [10_000, 20_000, 2_500],
    }
)

print("--- memeriksa transaksi_baik ---")
print("pelanggaran:", cek_schema(transaksi_baik, SCHEMA_TRANSAKSI) or "tidak ada (valid)")
print()
print("--- memeriksa transaksi_buruk ---")
for p in cek_schema(transaksi_buruk, SCHEMA_TRANSAKSI):
    print("-", p)
print()
print("--- gerbang menolak data yang melanggar ---")
try:
    gerbang_schema(transaksi_buruk, SCHEMA_TRANSAKSI)
except ValueError as exc:
    print("ValueError:", str(exc).splitlines()[0])
print()
print("--- data valid melewati gerbang ---")
lolos = gerbang_schema(transaksi_baik, SCHEMA_TRANSAKSI)
print("baris lolos:", len(lolos))

Output yang Dihasilkan:

text 
--- memeriksa transaksi_baik ---
pelanggaran: tidak ada (valid)

--- memeriksa transaksi_buruk ---
- kolom hilang: ['harga']
- dtype jumlah: float64 (harus Int64)
- jumlah memuat nilai negatif
- kode_transaksi duplikat: ['TRX-001']

--- gerbang menolak data yang melanggar ---
ValueError: schema tidak sesuai kontrak: kolom hilang: ['harga']; dtype jumlah: float64 (harus Int64); jumlah memuat nilai negatif; kode_transaksi duplikat: ['TRX-001']

--- data valid melewati gerbang ---
baris lolos: 3

Mari kita bedah bagian pentingnya:

  1. SCHEMA_TRANSAKSI: kontrak ditulis sebagai data, bukan kode percabangan. Menambah kolom wajib atau mengganti dtype harapan cukup mengubah dict ini — fungsi pemeriksa tidak perlu disentuh.
  2. Pemeriksaan kolom dan dtype: hilang dihitung dengan list comprehension terhadap df.columns, dan dtype dibandingkan langsung dengan string (df[kolom].dtype != "Int64"). transaksi_buruk memuat kolom harga_satuan yang tidak dikenal, sehingga tercatat sebagai kolom hilang: ['harga']; kolom jumlah berisi 2.5 sehingga dtype-nya float64 dan pelanggaran dtype jumlah: float64 (harus Int64) tercatat.
  3. Pemeriksaan isi: (df[kolom] < 0).any() menangkap nilai −1 pada kolom jumlah. Pemeriksaan ini sengaja dijalankan hanya bila kolomnya ada, sehingga satu laporan bisa memuat semua pelanggaran sekaligus, bukan berhenti di pelanggaran pertama.
  4. Pemeriksaan kunci unik: df["kode_transaksi"].is_unique bernilai False karena TRX-001 muncul dua kali, dan duplicated() menampilkan nilai mana yang duplikat: ['TRX-001']. Laporan yang menyebut nilainya mempercepat pemeriksaan ke sumber data.
  5. gerbang_schema(): bila daftar pelanggaran tidak kosong, fungsi melempar ValueError dengan seluruh pesan digabung. Dipanggil di awal pipeline, gerbang ini membuat data rusak berhenti di pintu masuk — bukan setelah tiga tahap transformasi menghabiskan waktu komputasi.
  6. Jalur sukses: transaksi_baik tanpa satu pun pelanggaran, dan gerbang mengembalikan DataFrame yang sama untuk diproses. Pola “kembalikan data yang lolos” membuat gerbang bisa disambungkan dengan pipe() ke rangkaian transformasi.

Aturan yang Tidak Tergantung Ukuran Data #

Semua pemeriksaan di atas berjalan sama baik pada tiga baris maupun tiga juta baris, karena berbasis operasi vektor — (df["jumlah"] < 0).any() hanya mengalokasikan satu nilai boolean. Artinya, gerbang schema boleh dipasang permanen tanpa memikirkan biaya, dan data sampel kecil yang dites di unit test memiliki aturan yang sama persis dengan data produksi.

Validator menjaga struktur data, tetapi tidak menjamin hasil transformasi benar. Untuk itu chapter berikutnya, Menguji Transformasi Data, mengunci perilaku tiap fungsi transformasi dengan input kecil, output terdefinisi, dan kasus tepi.

Referensi resmi untuk chapter ini: pandas.Series.is_unique , pandas.DataFrame.dtypes , dan Essential basic functionality .