Pemrosesan Ulang Data

Tiga bulan setelah pipeline penjualan berjalan, tim keuangan menemukan kejanggalan: laporan pendapatan masih menghitung pesanan yang sudah direfund. Bug tersebut bukan sekadar kesalahan hari ini — setiap angka sejak pipeline pertama kali dijalankan patut dicurigai, dan memperbaiki data besok saja tidak menutup laporan yang sudah terbit. Pemrosesan ulang (reprocess) adalah tindakan menjalankan kembali transformasi atas data historis, baik karena bug ditemukan, logika bisnis berubah, maupun data lama baru tersedia. Karena reprocess menimpa hasil yang mungkin sudah dikonsumsi, keselamatan proses tersebut tidak bisa diserahkan pada keberuntungan.

Tiga Alasan Reprocess #

Alasan paling umum adalah perbaikan bug: logika v1 keliru menghitung, sehingga seluruh hasilnya harus dihitung ulang dengan logika v2. Alasan kedua adalah logika baru: definisi metrik berubah — misalnya pendapatan kini wajib mengecualikan refund — dan manajemen ingin melihat angka lama dengan definisi baru. Alasan ketiga adalah backfill: sumber data baru atau partisi historis yang dulu tidak tersedia akhirnya bisa diakses, lalu periode lama perlu diisi. Ketiganya berbagi satu kebutuhan yang sama: menjalankan ulang kode atas data yang sudah pernah diproses, tanpa merusak apa pun di sepanjang jalan.

Syarat Aman: Idempotent dan Versioned Output #

Reprocess aman berdiri di atas dua syarat. Pertama, transformasi harus idempotent: menjalankan ulang partisi yang sama menghasilkan isi yang sama, sehingga kegagalan di tengah jalan cukup diatasi dengan menjalankan ulang dari awal (Chapter Idempotent Transformation). Kedua, hasil harus berversi: setiap baris output ditandai kode_versi transformasi yang menulisnya (atau ditulis ke tabel/berkas terpisah per versi). Penanda versi membuat campuran partisi lama dan baru terlihat jelas, memungkinkan audit siapa menulis apa, dan memberi jalan pemulihan bila v2 ternyata lebih buruk dari v1.

Implementasi: Reprocess Partisi dengan Kode v2 #

Tabel pesanan mencatat status tiap pesanan. Transform v1 menghitung seluruh pesanan sebagai pendapatan — bug-nya: pesanan refund ikut terhitung. Transform v2 mengecualikan refund, dan reprocess hanya menyentuh partisi 2026-01-10 yang memuat pesanan refund; partisi 2026-01-11 tidak berisi refund sehingga tidak perlu disentuh.

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,
    tanggal TEXT NOT NULL,
    pelanggan TEXT NOT NULL,
    produk TEXT NOT NULL,
    qty INTEGER NOT NULL,
    harga INTEGER NOT NULL,
    status TEXT NOT NULL
);
INSERT INTO pesanan VALUES
    (1, '2026-01-10', 'Budi', 'Kemeja Batik', 1, 250000, 'lunas'),
    (2, '2026-01-10', 'Siti', 'Kain Songket', 2, 180000, 'lunas'),
    (3, '2026-01-10', 'Dewi', 'Blouse Katun', 1, 120000, 'refund'),
    (4, '2026-01-10', 'Rina', 'Rok Tenun', 2, 150000, 'lunas'),
    (5, '2026-01-11', 'Budi', 'Seluar Linen', 1, 210000, 'lunas'),
    (6, '2026-01-11', 'Siti', 'Kemeja Batik', 3, 250000, 'lunas');
CREATE TABLE penjualan_harian (
    tanggal TEXT PRIMARY KEY,
    jumlah_pesanan INTEGER NOT NULL,
    total_pendapatan INTEGER NOT NULL,
    kode_versi TEXT NOT NULL
);
""")


def transform_partisi(koneksi, tanggal, versi_kode):
    koneksi.execute("DELETE FROM penjualan_harian WHERE tanggal = ?", (tanggal,))
    if versi_kode == 1:
        # v1: semua pesanan dihitung sebagai pendapatan (bug)
        koneksi.execute(
            """INSERT INTO penjualan_harian
               SELECT tanggal, COUNT(*), SUM(qty * harga), 'v1'
               FROM pesanan WHERE tanggal = ? GROUP BY tanggal""",
            (tanggal,),
        )
    else:
        # v2: pesanan refund dikeluarkan dari pendapatan
        koneksi.execute(
            """INSERT INTO penjualan_harian
               SELECT tanggal, COUNT(*), SUM(qty * harga), 'v2'
               FROM pesanan WHERE tanggal = ? AND status != 'refund'
               GROUP BY tanggal""",
            (tanggal,),
        )
    koneksi.commit()


def dump(koneksi):
    return koneksi.execute(
        "SELECT tanggal, jumlah_pesanan, total_pendapatan, kode_versi FROM penjualan_harian ORDER BY tanggal"
    ).fetchall()


print("=== Proses awal: kode v1 ===")
transform_partisi(koneksi, "2026-01-10", versi_kode=1)
transform_partisi(koneksi, "2026-01-11", versi_kode=1)
for baris in dump(koneksi):
    print(" ", baris)

print("=== Bug ditemukan: pesanan 3 (refund) ikut dihitung sebagai pendapatan ===")
print("=== Reprocess partisi 2026-01-10 saja dengan kode v2 ===")
transform_partisi(koneksi, "2026-01-10", versi_kode=2)
for baris in dump(koneksi):
    print(" ", baris)

print("=== Uji idempotensi: reprocess dijalankan dua kali ===")
transform_partisi(koneksi, "2026-01-10", versi_kode=2)
dump_1 = dump(koneksi)
transform_partisi(koneksi, "2026-01-10", versi_kode=2)
dump_2 = dump(koneksi)
print("  hasil identik:", dump_1 == dump_2)

print("=== Verifikasi: v2 partisi 2026-01-10 == agregat ulang langsung dari sumber ===")
benar = koneksi.execute(
    """SELECT COUNT(*), SUM(qty * harga) FROM pesanan
       WHERE tanggal = '2026-01-10' AND status != 'refund'"""
).fetchone()
(dump_10,) = koneksi.execute(
    "SELECT (jumlah_pesanan, total_pendapatan) = (?, ?) FROM penjualan_harian WHERE tanggal = '2026-01-10'",
    benar,
).fetchone()
print("  cocok:", bool(dump_10))

koneksi.close()

Output yang Dihasilkan:

=== Proses awal: kode v1 ===
  ('2026-01-10', 4, 1030000, 'v1')
  ('2026-01-11', 2, 960000, 'v1')
=== Bug ditemukan: pesanan 3 (refund) ikut dihitung sebagai pendapatan ===
=== Reprocess partisi 2026-01-10 saja dengan kode v2 ===
  ('2026-01-10', 3, 910000, 'v2')
  ('2026-01-11', 2, 960000, 'v1')
=== Uji idempotensi: reprocess dijalankan dua kali ===
  hasil identik: True
=== Verifikasi: v2 partisi 2026-01-10 == agregat ulang langsung dari sumber ===
  cocok: True

Mari kita telaah baris-baris kode di atas:

1. Baris 34-54 (transform_partisi): setiap pemanggilan menghapus partisi lalu mengisinya ulang dengan logika sesuai versi_kode; label 'v1'/'v2' ikut ditulis sebagai kolom kode_versi sehingga asal-usul tiap baris tercatat di data itu sendiri.
2. Baris 63-67 (proses awal): kedua partisi ditulis dengan v1 — angka 2026-01-10 tampak 4 pesanan Rp1030000 karena pesanan refund ikut terhitung.
3. Baris 70-73 (reprocess tertarget): hanya partisi 2026-01-10 dijalankan ulang dengan v2; hasilnya turun menjadi 3 pesanan Rp910000, sementara partisi 2026-01-11 tetap v1 karena isinya memang tidak terdampak bug.
4. Baris 75-91 (pembuktian ganda): reprocess dijalankan dua kali untuk membuktikan hasil identik (idempotent), lalu angka v2 dibandingkan langsung dengan agregat ulang dari sumber — kolom kode_versi yang campur (v1 dan v2) kini justru memetakan partisi mana yang sudah diperbaiki.

Full Rebuild vs Reprocess Partisi #

Pemetaan dampak tersebut bisa diotomasi: daripada menebak partisi mana yang terdampak, query sumber untuk mendaftarkannya — misalnya SELECT DISTINCT tanggal FROM pesanan WHERE status = 'refund' — lalu gunakan daftar tersebut sebagai parameter reprocess. Orchestrator seperti Airflow menyediakan mekanisme backfill yang menjalankan job per interval tanggal, sehingga reprocess tertarget tinggal menentukan rentang tanggalnya.

Di dbt, mekanisme yang sama tersedia sebagai bendera --full-refresh yang memaksa model incremental dibangun ulang dari nol, sedangkan eksekusi normal hanya memproses data baru — keduanya terdokumentasi di referensi perintah dbt .

Kita kini memegang kotak peralatan transformasi yang lengkap: incremental, idempotent, toleran terhadap data terlambat, dapat direkonsiliasi, tahan gagal, dan dapat diproses ulang. Pertanyaan terakhir section ini bersifat arsitektural: di mana sebaiknya transformasi berjalan — sebelum data masuk warehouse (ETL) atau setelahnya (ELT)? Chapter Memilih ETL atau ELT menutup dengan kerangka keputusan untuk menjawabnya.