Data Reconciliation

Job pemuatan melaporkan status sukses, tetapi “sukses” pada level job hanya berarti tidak ada exception yang terlempar — bukan berarti seluruh data berpindah dengan benar. Bug di query pemuatan bisa menghilangkan satu partisi penuh tanpa pesan galat apa pun, dan nilai yang ter-parse salah tetap lolos selama tipe datanya cocok. Data reconciliation adalah pemeriksaan independen yang membandingkan sumber dan target setelah pipeline selesai, sehingga selisih apa pun ketahuan sebelum dipakai tim analisis. Chapter Pemuatan Data menyoroti cara memindahkan data; bab ini memastikan hasil pemindahan tersebut dapat dipercaya.

Tiga Lapis Pemeriksaan #

Rekonsiliasi efisien bekerja dari yang paling murah ke yang paling teliti. Pemeriksaan jumlah baris menangkap data yang hilang atau berlebih secara kasar. Checksum agregat — jumlah pesanan dan total pendapatan per partisi — menangkap selisih nilai walaupun jumlah baris sama. Perbandingan sampel memeriksa sebagian baris detail untuk mendekati akar masalah; saat satu partisi terbukti bermasalah, drill-down membandingkan seluruh baris partisi tersebut.

Pemeriksaan	Yang ditangkap	Biaya
Jumlah baris per tabel/partisi	baris hilang atau berlebih	sangat murah
Checksum agregat per partisi	nilai berubah walau jumlah sama	murah
Perbandingan sampel baris detail	indikasi cepat kerusakan baris	sedang
Drill-down seluruh partisi bermasalah	baris dan kolom persis yang beda	mahal, terarah

Arah perbandingan juga perlu tegas: sumber (OLTP) selalu diperlakukan sebagai kebenaran, target diperiksa terhadap sumber — bukan sebaliknya. Checksum pada praktik produksi sering diperkuat dengan hash atas baris terurut (misalnya MD5 gabungan kolom) agar perubahan satu karakter pun terdeteksi, sesuatu yang penjumlahan saja bisa lewatkan bila dua selisih saling meniadakan.

Skenario: Pemuatan yang Diam-Diam Rusak #

Sistem OLTP toko kita punya tabel pesanan; di warehouse berdiri tabel pesanan_gudang beserta agregat penjualan_harian_gudang. Job pemuatan berjalan “sukses”, tetapi diam-diam membawa dua bug: partisi 2026-01-11 tidak ikut termuat, dan qty pesanan 9 ter-parse salah dari 2 menjadi 3. Kedua cacat tersebut tidak menggagalkan job — hanya rekonsiliasi yang bisa menangkapnya.

python
import random
import sqlite3
from pathlib import Path

DB = Path("toko.db")
if DB.exists():
    DB.unlink()

koneksi = sqlite3.connect(DB)
koneksi.executescript("""
CREATE TABLE pesanan (
    id INTEGER PRIMARY KEY,
    event_time TEXT NOT NULL,
    pelanggan TEXT NOT NULL,
    qty INTEGER NOT NULL,
    harga INTEGER NOT NULL
);
INSERT INTO pesanan VALUES
    (1, '2026-01-10T09:15:00', 'Budi', 1, 250000),
    (2, '2026-01-10T14:30:00', 'Siti', 2, 180000),
    (3, '2026-01-10T19:45:00', 'Dewi', 1, 120000),
    (4, '2026-01-11T08:05:00', 'Rina', 1, 210000),
    (5, '2026-01-11T13:20:00', 'Budi', 2, 150000),
    (6, '2026-01-11T16:40:00', 'Siti', 1, 250000),
    (7, '2026-01-12T07:40:00', 'Dewi', 2, 180000),
    (8, '2026-01-12T10:05:00', 'Budi', 1, 120000),
    (9, '2026-01-12T13:30:00', 'Rina', 2, 180000),
    (10, '2026-01-12T17:55:00', 'Siti', 1, 210000);
CREATE TABLE pesanan_gudang (
    id INTEGER PRIMARY KEY,
    event_time TEXT NOT NULL,
    pelanggan TEXT NOT NULL,
    qty INTEGER NOT NULL,
    harga INTEGER NOT NULL
);
CREATE TABLE penjualan_harian_gudang (
    tanggal TEXT PRIMARY KEY,
    jumlah_pesanan INTEGER NOT NULL,
    total_pendapatan INTEGER NOT NULL
);
""")

# Simulasi hasil job pemuatan bermasalah:
# bug 1: partisi 2026-01-11 (pesanan 4-6) tidak ikut termuat
# bug 2: qty pesanan 9 ter-parse salah menjadi 3
koneksi.executemany(
    "INSERT INTO pesanan_gudang VALUES (?, ?, ?, ?, ?)",
    [
        (1, "2026-01-10T09:15:00", "Budi", 1, 250000),
        (2, "2026-01-10T14:30:00", "Siti", 2, 180000),
        (3, "2026-01-10T19:45:00", "Dewi", 1, 120000),
        (7, "2026-01-12T07:40:00", "Dewi", 2, 180000),
        (8, "2026-01-12T10:05:00", "Budi", 1, 120000),
        (9, "2026-01-12T13:30:00", "Rina", 3, 180000),
        (10, "2026-01-12T17:55:00", "Siti", 1, 210000),
    ],
)
koneksi.execute("""
    INSERT INTO penjualan_harian_gudang
    SELECT substr(event_time, 1, 10), COUNT(*), SUM(qty * harga)
    FROM pesanan_gudang GROUP BY substr(event_time, 1, 10)
""")
koneksi.commit()


def rekonsiliasi_agregat():
    sumber = {
        t: (j, tot) for t, j, tot in koneksi.execute(
            """SELECT substr(event_time, 1, 10), COUNT(*), SUM(qty * harga)
               FROM pesanan GROUP BY substr(event_time, 1, 10)""")
    }
    target = {
        t: (j, tot) for t, j, tot in koneksi.execute(
            "SELECT tanggal, jumlah_pesanan, total_pendapatan FROM penjualan_harian_gudang")
    }
    (baris_sumber,) = koneksi.execute("SELECT COUNT(*) FROM pesanan").fetchone()
    (baris_target,) = koneksi.execute("SELECT COUNT(*) FROM pesanan_gudang").fetchone()
    print(f"  jumlah baris detail: sumber {baris_sumber} vs target {baris_target}")

    masalah = 0
    for tanggal in sorted(set(sumber) | set(target)):
        if tanggal not in target:
            print(f"  {tanggal}: TIDAK ADA di target (sumber: {sumber[tanggal][0]} pesanan, Rp{sumber[tanggal][1]})")
            masalah += 1
        elif tanggal not in sumber:
            print(f"  {tanggal}: kelebihan di target")
            masalah += 1
        else:
            (js, ts), (jt, tt) = sumber[tanggal], target[tanggal]
            if (js, ts) == (jt, tt):
                print(f"  {tanggal}: cocok ({js} pesanan, Rp{ts})")
            else:
                print(f"  {tanggal}: CHECKSUM BEDA -> sumber {js} pesanan/Rp{ts} vs target {jt} pesanan/Rp{tt}")
                masalah += 1
    print(f"  HASIL: {'LOLOS' if masalah == 0 else f'GAGAL ({masalah} masalah)'}")


def bandingkan_sampel(jumlah=3):
    pemilih = random.Random(42)
    semua_id = [r[0] for r in koneksi.execute("SELECT id FROM pesanan ORDER BY id")]
    for id_ in pemilih.sample(semua_id, jumlah):
        sumber = koneksi.execute(
            "SELECT pelanggan, qty, harga FROM pesanan WHERE id = ?", (id_,)).fetchone()
        target = koneksi.execute(
            "SELECT pelanggan, qty, harga FROM pesanan_gudang WHERE id = ?", (id_,)).fetchone()
        if sumber == target:
            print(f"  pesanan {id_}: cocok")
        elif target is None:
            print(f"  pesanan {id_}: hilang di target")
        else:
            print(f"  pesanan {id_}: sumber {sumber} vs target {target}")


def bandingkan_partisi(tanggal):
    for id_, pelanggan, qty, harga in koneksi.execute(
        """SELECT id, pelanggan, qty, harga FROM pesanan
           WHERE substr(event_time, 1, 10) = ? ORDER BY id""", (tanggal,)
    ):
        target = koneksi.execute(
            "SELECT pelanggan, qty, harga FROM pesanan_gudang WHERE id = ?", (id_,)
        ).fetchone()
        if target is None:
            print(f"  pesanan {id_}: hilang di target")
        elif (pelanggan, qty, harga) != target:
            print(f"  pesanan {id_}: sumber {(pelanggan, qty, harga)} vs target {target} -> akar masalah")


def muat_ulang_partisi(tanggal):
    koneksi.execute("DELETE FROM pesanan_gudang WHERE substr(event_time, 1, 10) = ?", (tanggal,))
    koneksi.execute(
        "INSERT INTO pesanan_gudang SELECT * FROM pesanan WHERE substr(event_time, 1, 10) = ?",
        (tanggal,))
    koneksi.execute("DELETE FROM penjualan_harian_gudang WHERE tanggal = ?", (tanggal,))
    koneksi.execute("""
        INSERT INTO penjualan_harian_gudang
        SELECT substr(event_time, 1, 10), COUNT(*), SUM(qty * harga)
        FROM pesanan_gudang WHERE substr(event_time, 1, 10) = ?
        GROUP BY substr(event_time, 1, 10)
    """, (tanggal,))
    koneksi.commit()


print("=== Rekonsiliasi agregat sumber vs target ===")
rekonsiliasi_agregat()

print("=== Bandingkan sampel 3 baris (seed 42) ===")
bandingkan_sampel()

print("=== Drill-down partisi 2026-01-12 (baris per baris) ===")
bandingkan_partisi("2026-01-12")

print("=== Muat ulang partisi bermasalah ===")
muat_ulang_partisi("2026-01-11")
muat_ulang_partisi("2026-01-12")

print("=== Rekonsiliasi ulang ===")
rekonsiliasi_agregat()

koneksi.close()

Output yang Dihasilkan:

text
=== Rekonsiliasi agregat sumber vs target ===
  jumlah baris detail: sumber 10 vs target 7
  2026-01-10: cocok (3 pesanan, Rp730000)
  2026-01-11: TIDAK ADA di target (sumber: 3 pesanan, Rp760000)
  2026-01-12: CHECKSUM BEDA -> sumber 4 pesanan/Rp1050000 vs target 4 pesanan/Rp1230000
  HASIL: GAGAL (2 masalah)
=== Bandingkan sampel 3 baris (seed 42) ===
  pesanan 2: cocok
  pesanan 1: cocok
  pesanan 5: hilang di target
=== Drill-down partisi 2026-01-12 (baris per baris) ===
  pesanan 9: sumber ('Rina', 2, 180000) vs target ('Rina', 3, 180000) -> akar masalah
=== Muat ulang partisi bermasalah ===
=== Rekonsiliasi ulang ===
  jumlah baris detail: sumber 10 vs target 10
  2026-01-10: cocok (3 pesanan, Rp730000)
  2026-01-11: cocok (3 pesanan, Rp760000)
  2026-01-12: cocok (4 pesanan, Rp1050000)
  HASIL: LOLOS

Mari kita telaah baris-baris kode di atas:

Baris 66-95 (rekonsiliasi_agregat): agregat sumber dan target dihitung ulang secara independen per tanggal, lalu setiap tanggal milik gabungan keduanya diperiksa — hilang di target, kelebihan, atau checksum (jumlah pesanan, total pendapatan) yang berbeda.
Baris 98-111 (bandingkan_sampel): tiga baris dipilih dengan random.Random(42) supaya sampel dapat direproduksi; setiap baris sumber dibandingkan kolom per kolom dengan baris target ber-id sama.
Baris 114-125 (bandingkan_partisi): drill-down membandingkan seluruh baris satu partisi; hanya baris yang hilang atau berbeda yang dicetak, sehingga pesanan 9 langsung tampak sebagai akar masalah.
Baris 128-140 (muat_ulang_partisi): perbaikan memakai pola rebuild partisi dari Chapter Idempotent Transformation — detail dan agregat partisi ditulis ulang dari sumber, lalu rekonsiliasi kedua memastikan LOLOS.

Hasilnya membaca seperti investigasi: pemeriksaan jumlah baris menunjukkan 10 lawan 7, checksum menandai partisi 2026-01-11 hilang dan 2026-01-12 bernilai salah, sampel menemukan pesanan 5 hilang, dan drill-down menunjuk pesanan 9 dengan qty 2 versus 3 sebagai biang kerusakan nilai. Perhatikan bahwa sampel tidak menemukan pesanan 9 — sampling adalah uji cepat, bukan bukti kelengkapan; drill-down pada partisi yang sudah terbukti bermasalah yang menutup kasus tersebut.

Agar rutin, rekonsiliasi perlu dibakukan sebagai langkah pipeline: jalankan setelah pemuatan selesai, simpan hasil beserta jumlah masalah ke tabel audit, dan picu peringatan bila angka tersebut lebih besar dari nol. Riwayat hasil tersebut juga menjadi alat diagnosis — lonjakan jumlah masalah pada tanggal tertentu hampir selalu berpasangan dengan perubahan kode atau skema yang terjadi di hari yang sama.

Rekonsiliasi bekerja pada data yang berhasil dimuat. Bagaimana dengan baris yang gagal di tengah transformasi — apakah seluruh batch harus dibatalkan? Chapter Penanganan Data yang Gagal memperkenalkan pola dead letter untuk mengkarantina baris bermasalah tanpa membatalkan sisanya.