Validasi Data saat Ingestion

Pipeline yang idempotent pada Chapter Idempotent Ingestion menjamin tidak ada baris ganda di warehouse, tetapi sama sekali tidak memeriksa kualitas baris yang masuk. Sumber pihak ketiga bisa berubah format tanpa pemberitahuan: kolom berganti nama, angka berubah menjadi teks, dan nilai absurd seperti total pesanan minus lima ribu rupiah lolos begitu saja. Sekali data rusak masuk ke warehouse, setiap laporan turunan mewarisi kerusakan tersebut — memperbaiki data di banyak tabel turunan jauh lebih mahal daripada menangkapnya di gerbang masuk. Chapter ini membahas dua lapis pemeriksaan, strategi fail fast versus karantina, dan membangun validator baris lengkap dengan berkas karantina.

Validasi Schema dan Validasi Nilai #

Pemeriksaan di gerbang ingestion bekerja pada dua lapis yang berbeda sasaran. Validasi schema memeriksa bentuk data: apakah kolom wajib ada, dan apakah tipe datanya sesuai kontrak. Validasi nilai memeriksa isi data: apakah angkanya masuk akal untuk domain bisnis kita.

Pemisahan ini penting karena keduanya menuntut perlakuan berbeda. Schema yang berubah total — misalnya kolom wajib menghilang — biasanya berarti sumber mengalami perubahan besar, sehingga seluruh proses sebaiknya dihentikan (fail fast) sebelum ada baris yang masuk. Sebaliknya, beberapa baris bernilai janggal dalam jutaan baris yang sehat adalah kondisi normal di dunia nyata; menghentikan seluruh pipeline karena lima baris buruk berarti menukar masalah kecil dengan pemadaman data.

Pola Karantina #

Pola karantina (quarantine) memberi baris bermasalah tempat tersendiri di luar tabel utama. Setiap baris yang gagal validasi ditulis ke berkas atau tabel karantina bersama daftar alasannya, lengkap dengan data aslinya — sehingga tidak ada informasi yang terbuang. Tim data bisa memeriksa isi karantina setiap hari, memperbaiki akar masalahnya di sumber, lalu memproses ulang baris yang sudah diperbaiki. Dengan cara ini, baris buruk tidak pernah menyentuh warehouse, namun juga tidak pernah menghilang tanpa jejak.

Contoh: Validator Baris dengan Karantina #

Kembali ke toko online kita. Tujuh baris pesanan tiba dari sumber, lima di antaranya menyimpan masalah yang berbeda-beda: total negatif, pelanggan null, kolom order_id hilang, total berupa teks, dan total di luar batas wajar. Validator memeriksa setiap baris, memisahkan yang sehat ke SQLite, dan mengarantina sisanya.

import json
import sqlite3
import tempfile
import os

BARIS_SUMBER = [
    {"order_id": "ORD-001", "pelanggan": "Budi", "total": 150000},
    {"order_id": "ORD-002", "pelanggan": "Siti", "total": -5000},
    {"order_id": "ORD-003", "pelanggan": None, "total": 99000},
    {"pelanggan": "Dewi", "total": 430000},
    {"order_id": "ORD-005", "pelanggan": "Budi", "total": "mahal"},
    {"order_id": "ORD-006", "pelanggan": "Siti", "total": 99999999},
    {"order_id": "ORD-007", "pelanggan": "Dewi", "total": 87000},
]

SKEMA = {"order_id": str, "pelanggan": str, "total": int}
RENTANG_TOTAL = (0, 50_000_000)


def validasi_baris(baris):
    galat = []
    for kolom, tipe in SKEMA.items():
        if kolom not in baris or baris[kolom] is None:
            galat.append(f"kolom {kolom} wajib ada dan tidak boleh null")
        elif not isinstance(baris[kolom], tipe):
            galat.append(f"kolom {kolom} harus {tipe.__name__}")
    if isinstance(baris.get("total"), int) and not (
        RENTANG_TOTAL[0] <= baris["total"] <= RENTANG_TOTAL[1]
    ):
        galat.append("total di luar rentang wajar")
    return galat


valid, karantina = [], []
for baris in BARIS_SUMBER:
    galat = validasi_baris(baris)
    if galat:
        karantina.append({"baris": baris, "alasan": galat})
    else:
        valid.append(baris)

db = sqlite3.connect(":memory:")
db.execute("""CREATE TABLE pesanan (
    order_id TEXT PRIMARY KEY,
    pelanggan TEXT NOT NULL,
    total INTEGER NOT NULL)""")
db.executemany(
    "INSERT INTO pesanan (order_id, pelanggan, total) VALUES (?, ?, ?)",
    [(b["order_id"], b["pelanggan"], b["total"]) for b in valid],
)
db.commit()

berkas_karantina = os.path.join(tempfile.mkdtemp(prefix="karantina-"), "karantina.jsonl")
with open(berkas_karantina, "w", encoding="utf-8") as f:
    for item in karantina:
        f.write(json.dumps(item, ensure_ascii=False) + "\n")

print(f"valid: {len(valid)} baris -> tabel pesanan")
print(f"gagal validasi: {len(karantina)} baris -> karantina.jsonl")
print("isi karantina.jsonl:")
with open(berkas_karantina, encoding="utf-8") as f:
    for baris in f:
        print(" ", baris.strip())
print("isi tabel pesanan:")
for baris in db.execute("SELECT order_id, pelanggan, total FROM pesanan ORDER BY order_id"):
    print(" ", tuple(baris))

Output yang Dihasilkan:

valid: 2 baris -> tabel pesanan
gagal validasi: 5 baris -> karantina.jsonl
isi karantina.jsonl:
  {"baris": {"order_id": "ORD-002", "pelanggan": "Siti", "total": -5000}, "alasan": ["total di luar rentang wajar"]}
  {"baris": {"order_id": "ORD-003", "pelanggan": null, "total": 99000}, "alasan": ["kolom pelanggan wajib ada dan tidak boleh null"]}
  {"baris": {"pelanggan": "Dewi", "total": 430000}, "alasan": ["kolom order_id wajib ada dan tidak boleh null"]}
  {"baris": {"order_id": "ORD-005", "pelanggan": "Budi", "total": "mahal"}, "alasan": ["kolom total harus int"]}
  {"baris": {"order_id": "ORD-006", "pelanggan": "Siti", "total": 99999999}, "alasan": ["total di luar rentang wajar"]}
isi tabel pesanan:
  ('ORD-001', 'Budi', 150000)
  ('ORD-007', 'Dewi', 87000)

Mari kita telaah baris-baris kode di atas:

1. Baris 16-17 (SKEMA dan RENTANG_TOTAL): aturan validasi dinyatakan sebagai data — kamus kolom-ke-tipe dan batas rentang — sehingga menambah kolom baru cukup menambah satu entri, bukan menulis ulang logika.
2. Baris 22-26 (validasi schema): perulangan memeriksa setiap kolom wajib; baris tanpa order_id dan baris dengan total bertipe teks tertangkap di lapis ini.
3. Baris 27-30 (validasi nilai): pemeriksaan rentang hanya dijalankan bila total memang berupa int — penjagaan isinstance ini mencegah perbandingan "mahal" <= 50_000_000 yang justru memunculkan TypeError.
4. Baris 34-40 (pemisahan): setiap baris diarahkan ke list valid atau karantina; baris karantina menyimpan data asli beserta daftar alasan, sehingga kegagalan bisa ditelusuri tanpa membuka ulang sumber.
5. Baris 53-56 (penulisan karantina): isi karantina disimpan sebagai JSONL — satu objek JSON per baris — agar mudah dibaca manusia sekaligus mudah diproses ulang oleh mesin setelah diperbaiki.

Perhatikan hasilnya pada output: hanya ORD-001 dan ORD-007 yang mencapai tabel pesanan, dan kelimanya tertangkap dengan alasan yang tepat. Validator seperti ini adalah versi mini dari kontrak data (data contract): kesepakatan eksplisit antara produsen dan konsumen data tentang bentuk dan makna setiap kolom.

Alat Validasi di Ekosistem #

Untuk pipeline besar, aturan validasi tidak lagi ditulis sebagai fungsi buatan sendiri. Great Expectations menyediakan kerangka expectation yang bisa dijalankan sebagai checkpoint berkala di https://docs.greatexpectations.io/ , sementara Pandera memvalidasi DataFrame pandas dan pyarrow dengan deklarasi schema di https://pandera.readthedocs.io/ . Keduanya memakai ide yang sama dengan contoh kita: aturan sebagai data, hasil sebagai laporan yang bisa ditindaklanjuti.

Validasi menutup pintu bagi data yang buruk, namun proses ingestion itu sendiri tetap bisa gagal di tengah jalan — koneksi putus, API mati, atau disk penuh. Klasifikasi kegagalan dan strategi pemulihannya menjadi pokok bahasan terakhir section ini pada Chapter Menangani Kegagalan Ingestion.