Bagian 3: Memvalidasi Hasil Transformasi

Bagian 2 berakhir dengan klaim: data bersih berisi 19 baris, sembilan kolom, dan omzet 2261000. Klaim seperti itu belum berguna bagi tim keuangan sebelum ada pemeriksaan yang bisa dijalankan ulang kapan pun. Pada bagian ini kita mengubah klaim menjadi gerbang validasi: tiga fungsi validator yang menguji skema, rentang nilai, dan keunikan kunci, ditambah satu perbandingan assert_frame_equal() terhadap ekspektasi beku hasil verifikasi Bagian 2. Setiap hasil pemeriksaan dicatat lewat logging deterministik ke stdout tanpa timestamp, pola yang dibahas pada chapter Logging dan Penanganan Error di section Praktik Data Engineering, dengan konsep pengujian dari section Pengujian dan Debugging.

Empat Gerbang Validasi #

Validator ditulis sebagai fungsi kecil yang melempar ValueError dengan pesan deterministik bila pemeriksaan gagal; pipeline berhenti sebelum data yang tidak konsisten menyebar ke tahap ekspor.

Gerbang	Fungsi	Gagal bila
`validasi_skema()`	kolom wajib ada, dtype sesuai	ada kolom hilang atau dtype bergeser
`validasi_rentang()`	nilai bermakna dan konsisten	`qty`/`harga_satuan` nonpositif, `total` tidak cocok, tanggal di luar periode
`validasi_unik()`	`id_transaksi` menjadi unique key	ada id ganda yang lolos deduplikasi
ekspektasi beku	`assert_frame_equal()` vs 19 baris	hasil pipeline berubah sedikit saja

Gerbang pertama membandingkan df.columns dengan daftar KOLOM_WAJIB, lalu memeriksa tipe setiap kolom terhadap peta TIPE_HARAPAN: str untuk teks, datetime64[us] untuk tanggal, dan int64 untuk kolom angka. Gerbang kedua menjalankan tiga uji nilai sekaligus — kepositifan qty dan harga_satuan, kesamaan total dengan qty * harga_satuan, dan Series.between("2026-01-01", "2026-03-31") pada kolom tanggal. Gerbang ketiga memanfaatkan properti Series.is_unique; bila gagal, daftar id yang ganda ikut dilaporkan agar penyebabnya mudah ditelusuri.

Ekspektasi Beku sebagai Standar Emas #

Fungsi ekspektasi() mengembalikan DataFrame 19 baris yang dibangun dari literal Python: id, tanggal, produk, kategori, kota, qty, harga_satuan, total, dan sumber. Angka-angkanya adalah hasil Bagian 2 yang sudah diperiksa manusia, lalu dibekukan di dalam kode. Kolom tanggal dikonversi dengan pd.to_datetime() saat ekspektasi dibangun agar dtype-nya sama persis dengan hasil pipeline. Standar emas seperti ini bukan berarti beku selamanya: bila aturan bisnis berubah — misalnya transaksi baru masuk — ekspektasi diperbarui lewat keputusan sadar, bukan diam-diam oleh hasil program.

Logging Tanpa Timestamp agar Dapat Dibandingkan #

Konfigurasi logger di awal skrip memakai logging.getLogger("etl") dengan StreamHandler(sys.stdout) dan Formatter("%(levelname)s:%(name)s:%(message)s"). Dua keputusan di sini penting. Pertama, tanpa timestamp, keluaran dua eksekusi yang berbeda bisa dibandingkan byte demi byte — syarat pipeline yang dapat direproduksi. Kedua, handler menunjuk sys.stdout secara eksplisit karena logging.basicConfig() bawaan menulis ke stderr dan menyisipkan waktu. Logger anak seperti etl.skema mewarisi handler induknya, sehingga tiap gerbang punya nama sendiri di keluaran.

python

"""Gerbang validasi hasil pipeline: skema, rentang, keunikan, dan ekspektasi beku."""
from __future__ import annotations

import logging
import sys
import tempfile
from pathlib import Path

import pandas as pd
from pandas.testing import assert_frame_equal

from data_sumber import muat_csv, muat_json, muat_sqlite

# Logger deterministik: keluaran ke stdout, format tanpa timestamp.
logger = logging.getLogger("etl")
logger.setLevel(logging.INFO)
logger.propagate = False
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s"))
logger.addHandler(handler)
log_skema = logging.getLogger("etl.skema")
log_rentang = logging.getLogger("etl.rentang")
log_unik = logging.getLogger("etl.unik")
log_gerbang = logging.getLogger("etl.ekspektasi")

KOLOM_WAJIB = [
    "id_transaksi", "tanggal", "produk", "kategori", "kota",
    "qty", "harga_satuan", "total", "sumber",
]
TIPE_HARAPAN = {
    "id_transaksi": "str",
    "tanggal": "datetime64[us]",
    "produk": "str",
    "kategori": "str",
    "kota": "str",
    "qty": "int64",
    "harga_satuan": "int64",
    "total": "int64",
    "sumber": "str",
}


def transformasi(
    csv_raw: pd.DataFrame, json_raw: pd.DataFrame, sqlite_raw: pd.DataFrame
) -> pd.DataFrame:
    """Seluruh tahap transformasi Bagian 2, tanpa pencetakan."""
    csv_df = csv_raw.rename(columns={"kode_trx": "id_transaksi"})
    csv_df["tanggal"] = pd.to_datetime(csv_df["tanggal"], format="%d/%m/%Y")
    csv_df["kota"] = csv_df["kota"].str.strip().str.title()
    csv_df = csv_df[csv_df["harga_satuan"] > 0]
    csv_df["sumber"] = "laporan_csv"

    json_df = json_raw.rename(
        columns={
            "order_id": "id_transaksi",
            "item": "produk",
            "kota_tujuan": "kota",
            "jml_barang": "qty",
        }
    )
    json_df = json_df[json_df["qty"] > 0]
    json_df["harga_satuan"] = json_df["harga_total"] // json_df["qty"]
    json_df["tanggal"] = pd.to_datetime(json_df["created_at"]).dt.normalize()
    json_df["kota"] = json_df["kota"].str.strip().str.title()
    json_df = json_df.drop(columns=["created_at", "harga_total"])
    json_df["sumber"] = "marketplace_json"

    sqlite_df = sqlite_raw.rename(
        columns={
            "trx_id": "id_transaksi",
            "tgl_trx": "tanggal",
            "nama_barang": "produk",
            "wilayah": "kota",
            "jml": "qty",
            "harga": "harga_satuan",
        }
    )
    sqlite_df["tanggal"] = pd.to_datetime(sqlite_df["tanggal"])
    sqlite_df = sqlite_df.drop(columns="kasir")
    sqlite_df["sumber"] = "kasir_sqlite"

    kolom_inti = [
        "id_transaksi", "tanggal", "produk", "kategori", "kota", "qty", "harga_satuan",
    ]
    gabung = pd.concat(
        [
            csv_df[kolom_inti + ["sumber"]],
            json_df[kolom_inti + ["sumber"]],
            sqlite_df[kolom_inti + ["sumber"]],
        ],
        ignore_index=True,
    )
    bersih = gabung.drop_duplicates(subset="id_transaksi", keep="last")
    bersih = bersih.reset_index(drop=True)
    bersih["total"] = bersih["qty"] * bersih["harga_satuan"]
    return bersih[
        ["id_transaksi", "tanggal", "produk", "kategori", "kota",
         "qty", "harga_satuan", "total", "sumber"]
    ]


def ekspektasi() -> pd.DataFrame:
    """Ekspektasi beku: 19 baris hasil pipeline yang sudah diverifikasi."""
    return pd.DataFrame(
        {
            "id_transaksi": [
                "TRX-A-001", "TRX-A-002", "TRX-A-003", "TRX-A-005", "TRX-A-006",
                "WEB-001", "WEB-002", "WEB-003", "WEB-004", "WEB-006", "WEB-007",
                "WEB-008", "TRX-B-001", "TRX-B-002", "TRX-B-003", "TRX-B-004",
                "TRX-B-005", "TRX-B-006", "TRX-B-007",
            ],
            "tanggal": pd.to_datetime(
                [
                    "2026-01-14", "2026-01-16", "2026-01-21", "2026-02-18", "2026-02-21",
                    "2026-01-03", "2026-01-05", "2026-01-09", "2026-02-02", "2026-02-14",
                    "2026-03-04", "2026-03-11", "2026-01-15", "2026-01-20", "2026-02-17",
                    "2026-02-19", "2026-02-23", "2026-03-06", "2026-03-15",
                ]
            ),
            "produk": [
                "Kopi Robusta Bubuk 200g", "Gula Pasir 1kg", "Beras Pandan Wangi 5kg",
                "Teh Celup Melati 25", "Mie Goreng Instan 85g", "Kopi Robusta Bubuk 200g",
                "Gula Pasir 1kg", "Mie Goreng Instan 85g", "Teh Celup Melati 25",
                "Gula Pasir 1kg", "Minyak Goreng 2L", "Kopi Robusta Bubuk 200g",
                "Kopi Robusta Bubuk 200g", "Beras Pandan Wangi 5kg", "Minyak Goreng 2L",
                "Teh Celup Melati 25", "Gula Pasir 1kg", "Mie Goreng Instan 85g",
                "Kopi Robusta Bubuk 200g",
            ],
            "kategori": [
                "Minuman", "Bahan Pokok", "Bahan Pokok", "Minuman", "Bahan Pokok",
                "Minuman", "Bahan Pokok", "Bahan Pokok", "Minuman", "Bahan Pokok",
                "Bahan Pokok", "Minuman", "Minuman", "Bahan Pokok", "Bahan Pokok",
                "Minuman", "Bahan Pokok", "Bahan Pokok", "Minuman",
            ],
            "kota": [
                "Bandung", "Bandung", "Bandung", "Bandung", "Bandung", "Jakarta",
                "Jakarta", "Bekasi", "Bogor", "Jakarta", "Bekasi", "Depok", "Denpasar",
                "Denpasar", "Denpasar", "Denpasar", "Denpasar", "Denpasar", "Denpasar",
            ],
            "qty": [4, 2, 6, 5, 5, 4, 4, 1, 5, 5, 2, 3, 6, 5, 3, 5, 3, 2, 5],
            "harga_satuan": [
                38500, 16000, 72000, 12500, 3500, 38500, 16000, 3500, 12500, 16000,
                36500, 38500, 38500, 72000, 36500, 12500, 16000, 3500, 38500,
            ],
            "total": [
                154000, 32000, 432000, 62500, 17500, 154000, 64000, 3500, 62500,
                80000, 73000, 115500, 231000, 360000, 109500, 62500, 48000, 7000,
                192500,
            ],
            "sumber": [
                "laporan_csv", "laporan_csv", "laporan_csv", "laporan_csv",
                "laporan_csv", "marketplace_json", "marketplace_json",
                "marketplace_json", "marketplace_json", "marketplace_json",
                "marketplace_json", "marketplace_json", "kasir_sqlite",
                "kasir_sqlite", "kasir_sqlite", "kasir_sqlite", "kasir_sqlite",
                "kasir_sqlite", "kasir_sqlite",
            ],
        }
    )


def validasi_skema(df: pd.DataFrame) -> None:
    """Gerbang 1: kolom wajib ada dan tipe data sesuai."""
    hilang = [kolom for kolom in KOLOM_WAJIB if kolom not in df.columns]
    if hilang:
        raise ValueError(f"kolom wajib tidak lengkap: {hilang}")
    log_skema.info("9 kolom wajib lengkap")
    salah = {
        kolom: str(df[kolom].dtype)
        for kolom, tipe in TIPE_HARAPAN.items()
        if str(df[kolom].dtype) != tipe
    }
    if salah:
        raise ValueError(f"tipe data tidak sesuai: {salah}")
    log_skema.info("tipe data sesuai (str, datetime64[us], int64)")


def validasi_rentang(df: pd.DataFrame) -> None:
    """Gerbang 2: nilai berada pada rentang yang bermakna."""
    if (df["qty"] <= 0).any() or (df["harga_satuan"] <= 0).any():
        raise ValueError("ada qty atau harga_satuan bernilai 0 atau negatif")
    log_rentang.info("qty dan harga_satuan positif di semua baris")
    if not (df["total"] == df["qty"] * df["harga_satuan"]).all():
        raise ValueError("total tidak sama dengan qty * harga_satuan")
    log_rentang.info("total = qty * harga_satuan di semua baris")
    if not df["tanggal"].between("2026-01-01", "2026-03-31").all():
        raise ValueError("ada tanggal di luar periode Januari-Maret 2026")
    log_rentang.info("tanggal berada pada rentang 2026-01-01..2026-03-31")


def validasi_unik(df: pd.DataFrame) -> None:
    """Gerbang 3: id_transaksi menjadi unique key."""
    if not df["id_transaksi"].is_unique:
        ganda = df["id_transaksi"][df["id_transaksi"].duplicated()].tolist()
        raise ValueError(f"id_transaksi ganda: {ganda}")
    log_unik.info("id_transaksi unik di %d baris", len(df))


def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        folder = Path(tmp)
        csv_raw = muat_csv(folder)
        json_raw = muat_json()
        sqlite_raw = muat_sqlite(folder)
    hasil = transformasi(csv_raw, json_raw, sqlite_raw)

    validasi_skema(hasil)
    validasi_rentang(hasil)
    validasi_unik(hasil)

    try:
        assert_frame_equal(hasil, ekspektasi())
    except AssertionError:
        log_gerbang.error("hasil pipeline berbeda dari ekspektasi beku")
        raise
    log_gerbang.info("assert_frame_equal lolos terhadap %d baris ekspektasi beku", len(hasil))

    print()
    print("Semua gerbang validasi lolos: hasil pipeline identik dengan ekspektasi beku.")


if __name__ == "__main__":
    main()

Output yang Dihasilkan:

text
INFO:etl.skema:9 kolom wajib lengkap
INFO:etl.skema:tipe data sesuai (str, datetime64[us], int64)
INFO:etl.rentang:qty dan harga_satuan positif di semua baris
INFO:etl.rentang:total = qty * harga_satuan di semua baris
INFO:etl.rentang:tanggal berada pada rentang 2026-01-01..2026-03-31
INFO:etl.unik:id_transaksi unik di 19 baris
INFO:etl.ekspektasi:assert_frame_equal lolos terhadap 19 baris ekspektasi beku

Semua gerbang validasi lolos: hasil pipeline identik dengan ekspektasi beku.

Mari kita telaah bagian-bagian pentingnya:

transformasi(): seluruh langkah Bagian 2 dikemas menjadi fungsi murni tanpa pencetakan, sehingga skrip ini tetap mandiri — memuat sumber, mentransformasi, lalu memvalidasi dalam satu eksekusi.
Tujuh baris INFO: tiap pemeriksaan yang lolos dicatat satu baris dengan nama gerbangnya, dari etl.skema sampai etl.ekspektasi, tanpa satu pun angka yang berubah antar eksekusi.
assert_frame_equal(hasil, ekspektasi()): perbandingan ketat terhadap nilai, dtype, dan index; bila lolos, satu baris log menyatakan 19 baris ekspektasi terpenuhi, lalu print() menutup dengan ringkasan.

Jembatan ke Ekspor #

Dengan empat gerbang lolos, dataset 19 baris kini punya bukti kualitas yang dapat ditunjukkan. Bagian 4 menutup proyek: mengekspor data bersih ke Parquet, ringkasan omzet per kota ke CSV, dan tabel SQLite untuk aplikasi lain, lengkap dengan rekap aliran baris dari setiap sumber.

Sumber bacaan resmi: pandas.testing.assert_frame_equal , pandas.Series.is_unique , dan pandas.Series.between .