Recovery dan Reprocessing

Hotfix untuk bug transform sudah dideploy pukul 05:00, tetapi pekerjaan baru dimulai: tabel hasil masih menyimpan angka salah untuk dua partisi November, dan agregat di hilir mart mingguan turut tercemar. Pertanyaan berikutnya bukan lagi soal kode, melainkan soal strategi: tabel mana saja yang harus diproses ulang, dalam urutan apa, berapa lama, dan siapa saja konsumen yang wajib menerima angka terkoreksi. Chapter ini membahas strategi reprocessing — melengkapi Chapter Pipeline Recovery yang membahas mekanika checkpoint dan resume untuk run yang gagal di tengah jalan.

Backfill, Replay, dan Dual-Write #

Tiga istilah ini sering dipakai bergantian padahal menjawab situasi berbeda. Backfill memproses ulang rentang historis dengan kode yang sudah benar untuk mengisi ulang tabel. Replay mengulang event dari titik simpan — offset topic Kafka atau antrean pesan — dalam urutan asli. Dual-write menulis dua versi output berdampingan selama masa migrasi, seperti pola tabel pesanan_v1/pesanan_v2 pada Chapter Pipeline Versioning:

Aspek	Backfill	Replay	Dual-write
Definisi	proses ulang rentang historis dengan kode saat ini	ulang event dari offset/antrean, urutan asli	tulis dua output berdampingan selama migrasi
Kapan dipakai	bug transform diperbaiki; logika baru diterapkan ke historis	pipeline streaming berhenti; event hilang setelah offset tertentu	ganti skema output; validasi versi baru sebelum switch
Sumber data	salinan raw di storage	log event yang masih tersimpan	sumber sama, dua jalur penulisan
Risiko utama	durasi panjang; hasil sementara dibaca konsumen	duplikasi bila offset tidak presisi	biaya compute ganda; dua sumber kebenaran sementara

Untuk kasus TokoKita, jawabannya adalah backfill terbatas: hanya partisi yang ditulis oleh kode bermasalah. Orkestrator seperti Apache Airflow menyediakan konsep backfill untuk memicu ulang run pada rentang tanggal tertentu — mekanismenya ada di dokumentasi Apache Airflow .

Blast Radius: Menelusuri Tabel Terdampak #

Blast radius adalah himpunan semua tabel dan konsumen yang tersentuh data salah. Penelusurannya mengikuti lineage ke bawah — materi Chapter Data Lineage: dari tabel yang salah, ikuti setiap dependen sampai ke dashboard dan laporan. Untuk kasus fct_pesanan partisi 12-13 November:

flowchart TD A["fct_pesanan partisi 12-13 Nov: proses ulang"] --> B["mart_penjualan_harian: lewati, full refresh otomatis"] A --> C["mart_gmv_mingguan: proses ulang 2026-W46"] B --> D["dashboard_gmv: verifikasi pasca-republikasi"] B --> E["laporan_finance: kirim notice koreksi"] C --> D

Dua tabel hilir menuntut perlakuan berbeda. mart_penjualan_harian bersifat full refresh — dibangun ulang penuh setiap run terjadwal, jadi cukup dipastikan run malam berikutnya berjalan setelah fct_pesanan bersih. mart_gmv_mingguan bersifat incremental, sehingga partisi minggu 2026-W46 harus diproses ulang secara eksplisit. Sifat penulisan tiap tabel inilah yang menentukan mana yang bisa dilewati.

Perencana Reprocessing dari Run Ledger #

Daftar partisi terdampak tidak perlu ditebak: tabel run_metadata dari Chapter Pipeline Versioning sudah mencatat versi kode untuk setiap run. Skrip berikut membaca run ledger, menemukan partisi yang ditulis kode v1.1.0 (versi dengan bug diskon ganda), menelusuri lineage, lalu menyusun rencana lengkap dengan durasi dari riwayat run:

python
"""Perencana reprocessing: dari run ledger (partisi yang ditulis kode bermasalah)
dan lineage tabel, susun urutan pemrosesan ulang, tandai yang bisa dilewati,
lalu hitung durasi dan blast radius."""
from datetime import date

RUN_LEDGER = [
    # (task, partisi, kode_versi, status, durasi_menit) -- dari tabel run_metadata
    ("fct_pesanan", "2026-11-10", "v1.0.0", "sukses", 24),
    ("fct_pesanan", "2026-11-11", "v1.0.0", "sukses", 25),
    ("fct_pesanan", "2026-11-12", "v1.1.0", "sukses", 25),
    ("fct_pesanan", "2026-11-13", "v1.1.0", "sukses", 25),
    ("mart_gmv_mingguan", "2026-W45", "v1.0.0", "sukses", 14),
    ("mart_gmv_mingguan", "2026-W46", "v1.1.0", "sukses", 15),
    ("mart_penjualan_harian", "2026-11-13", "v1.1.0", "sukses", 40),
]

TABEL_BUG = "fct_pesanan"   # tabel yang transform-nya salah
KODE_BUG = "v1.1.0"         # versi kode yang menghitung diskon dua kali

LINEAGE = {
    "fct_pesanan": ["mart_penjualan_harian", "mart_gmv_mingguan"],
    "mart_penjualan_harian": ["dashboard_gmv", "laporan_finance"],
    "mart_gmv_mingguan": ["dashboard_gmv"],
    "dashboard_gmv": [],
    "laporan_finance": [],
}

SIFAT = {  # pola penulisan tiap tabel menentukan aksi reprocessing
    "fct_pesanan": "incremental",          # INSERT OR REPLACE per partisi harian
    "mart_penjualan_harian": "full_refresh",  # dibangun ulang penuh tiap run
    "mart_gmv_mingguan": "incremental",    # upsert per partisi mingguan
    "dashboard_gmv": "passive",            # hanya membaca, tidak menyimpan
    "laporan_finance": "passive",
}


def partisi_mingguan(tanggal):
    """Partisi minggu ISO yang memuat tanggal, mis. 2026-11-12 -> 2026-W46."""
    d = date.fromisoformat(tanggal)
    iso = d.isocalendar()
    return f"{iso.year}-W{iso.week:02d}"


def durasi_ledger(task, partisi):
    """Ambil durasi historis tercatat untuk pasangan task-partisi."""
    for t, p, _, _, menit in RUN_LEDGER:
        if t == task and p == partisi:
            return menit
    return 15  # perkiraan konservatif bila belum ada riwayat


# 1) partisi terdampak: semua run tabel bermasalah yang ditulis kode bermasalah
partisi_bug = sorted(p for t, p, v, _, _ in RUN_LEDGER
                     if t == TABEL_BUG and v == KODE_BUG)

# 2) blast radius: telusuri lineage ke bawah (BFS) dari tabel bermasalah
blast_radius, antre = [], [TABEL_BUG]
while antre:
    for bawah in LINEAGE[antre.pop(0)]:
        if bawah not in blast_radius:
            blast_radius.append(bawah)
            antre.append(bawah)

# 3) susun rencana dalam urutan topologis (level demi level dari BFS)
rencana, total = [], 0
for task in [TABEL_BUG] + blast_radius:
    if task == TABEL_BUG:
        partisi = partisi_bug
    elif SIFAT[task] == "incremental":
        partisi = sorted({partisi_mingguan(p) for p in partisi_bug})
    else:
        partisi = ["-"]
    if task == TABEL_BUG or SIFAT[task] == "incremental":
        aksi = "proses ulang"
        est = sum(durasi_ledger(task, p) for p in partisi)
        total += est
    elif SIFAT[task] == "full_refresh":
        aksi, est = "lewati: rebuild otomatis pada run terjadwal", 0
    else:
        aksi, est = "verifikasi pasca-republikasi + notice", 10
        total += 10
    rencana.append((task, aksi, partisi, est))

print(f"Partisi terdampak ({TABEL_BUG}, kode {KODE_BUG}): {', '.join(partisi_bug)}")
print(f"Blast radius: {', '.join(blast_radius)}\n")
print("| # | Tabel | Aksi | Partisi | Durasi est. |")
print("|---|---|---|---|---|")
for i, (task, aksi, partisi, est) in enumerate(rencana, 1):
    print(f"| {i} | {task} | {aksi} | {', '.join(partisi)} | "
          f"{'otomatis' if est == 0 else f'{est} m'} |")

konsumen = [t for t in blast_radius if SIFAT[t] == "passive"]
print(f"\nKonsumen akhir yang wajib menerima notice: {', '.join(konsumen)}")
print(f"Durasi total rencana: {total} menit ({total / 60:.1f} jam)")

Output yang Dihasilkan:

text
Partisi terdampak (fct_pesanan, kode v1.1.0): 2026-11-12, 2026-11-13
Blast radius: mart_penjualan_harian, mart_gmv_mingguan, dashboard_gmv, laporan_finance

| # | Tabel | Aksi | Partisi | Durasi est. |
|---|---|---|---|---|
| 1 | fct_pesanan | proses ulang | 2026-11-12, 2026-11-13 | 50 m |
| 2 | mart_penjualan_harian | lewati: rebuild otomatis pada run terjadwal | - | otomatis |
| 3 | mart_gmv_mingguan | proses ulang | 2026-W46 | 15 m |
| 4 | dashboard_gmv | verifikasi pasca-republikasi + notice | - | 10 m |
| 5 | laporan_finance | verifikasi pasca-republikasi + notice | - | 10 m |

Konsumen akhir yang wajib menerima notice: dashboard_gmv, laporan_finance
Durasi total rencana: 85 menit (1.4 jam)

Mari kita telaah baris-baris kode di atas:

Baris 33-34 (partisi_bug): partisi terdampak ditemukan dengan memfilter run ledger pada tabel bermasalah dan versi kode bermasalah — bukan dengan menebak tanggal dari ingatan.
Baris 36-42 (BFS lineage): penelusuran ke bawah mengumpulkan blast radius level demi level; hasil traversal sekaligus menjadi urutan eksekusi yang menghormati dependensi.
Baris 49-50 (partisi_mingguan): tanggal harian dipetakan ke partisi minggu ISO, sehingga tabel hilir dengan granularitas berbeda tetap bisa dijadwalkan.
Baris 60-71 (keputusan aksi): sifat penulisan tabel menentukan aksi — incremental diproses ulang eksplisit, full refresh dilewati karena rebuild otomatis, passive cukup diverifikasi.
Baris 63 (durasi_ledger): estimasi durasi memakai durasi nyata dari run sebelumnya, bukan angka dari udara.

Mengoreksi Angka yang Sudah Dibaca Konsumen #

Bagian tersulit dari reprocessing bukan teknis, melainkan komunikasi: angka salah mungkin sudah diunduh, dipresentasikan, atau dipakai keputusan. Tiga aturan mainnya: umumkan sebelum mulai memproses ulang (agar tidak ada yang menerbitkan angka lama saat transisi), cantumkan angka lama dan angka baru berdampingan, dan beri stempel waktu kapan angka final tayang. Contoh notice yang dikirim ke kanal #finance:

text
Pemberitahuan Koreksi Data — 2026-11-14 14:30
Tabel: mart_gmv_mingguan, partisi 2026-W46
Kejadian: bug transform menghitung diskon dua kali pada pesanan 12-13 November.
Dampak: GMV W46 tayang Rp61.480.000; angka yang benar Rp63.920.000.
Tindakan: partisi diproses ulang pukul 13:10-14:25; angka final tayang 14:30.
Laporan yang diunduh sebelum pukul 14:30 mohon diunduh ulang.
Kontak: Dewi (tim platform data)

Notice semacam ini adalah kelanjutan alami komunikasi insiden dari Chapter Incident Response: timeline insiden mencatat kapan bug diperbaiki, sedangkan notice koreksi menutup sisi konsumen datanya. Satu pertanyaan pencegahan tetap tersisa: mengapa versi kode bermasalah sampai dipromosikan ke production, dan bagaimana cara memutar balikannya dengan disiplin? Chapter Versioning dan Rollback menjawab keduanya.