Penanganan Data yang Gagal

Mitra toko online kita mengirim berkas batch berisi ribuan pesanan setiap malam, dan hampir setiap bulan ada satu-dua baris yang rusak: JSON terpotong di tengah string, kolom hilang, atau nilai negatif yang tidak masuk akal. Bila satu baris rusak membuat seluruh batch dibatalkan, 4.998 pesanan sehat ikut tertahan demi dua baris bermasalah — laporan pagi telat dan tim operasional menunggu. Pola dead letter membalik kebijakan tersebut: baris yang gagal dikarantina beserta bukti penyebabnya, sisanya diproses normal, dan baris karantina diproses ulang setelah akar masalahnya ditangani.

Pola Dead Letter dan Isolasi Penyebab #

Dead letter queue (DLQ) adalah tabel atau antrean khusus yang menampung baris gagal beserta payload mentah, pesan galat, klasifikasi penyebab, dan jumlah percobaan. Dengan catatan tersebut, kegagalan bisa diisolasi menjadi dua kelas yang jalur perbaikannya berbeda: kelas data berarti payload memang cacat sehingga sumber harus memperbaiki lalu mengirim ulang, sedangkan kelas kode berarti transform tidak siap menghadapi data yang sah sehingga logika transform yang harus diperbaiki. Dua kelas tersebut menentukan siapa yang bertindak: tim pemilik data sumber atau tim pipeline.

Implementasi: Batch dengan Baris Korup #

Skrip berikut memproses sepuluh baris JSONL. Enam baris sehat, tiga baris membawa cacat data (JSON terpotong, kolom qty hilang, qty negatif), dan satu baris berisi diskon bernilai null — data yang sah, tetapi transform v1 tidak siap sehingga galatnya jatuh ke kelas kode.

python
import json
import sqlite3
from pathlib import Path

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

koneksi = sqlite3.connect(DB)
koneksi.executescript("""
CREATE TABLE pesanan (
    pesanan_id INTEGER PRIMARY KEY,
    waktu TEXT NOT NULL,
    pelanggan TEXT NOT NULL,
    produk TEXT NOT NULL,
    qty INTEGER NOT NULL,
    harga INTEGER NOT NULL,
    total INTEGER NOT NULL
);
CREATE TABLE dead_letter (
    dl_id INTEGER PRIMARY KEY AUTOINCREMENT,
    baris INTEGER NOT NULL,
    payload TEXT NOT NULL,
    galat TEXT NOT NULL,
    kelas TEXT NOT NULL,
    percobaan INTEGER NOT NULL DEFAULT 1
);
""")

WAJIB = ("pesanan_id", "waktu", "pelanggan", "produk", "qty", "harga")


def transform(payload, versi_kode):
    data = json.loads(payload)
    for kolom in WAJIB:
        if kolom not in data:
            raise ValueError(f"kolom wajib hilang: {kolom}")
    if not isinstance(data["qty"], int) or data["qty"] <= 0:
        raise ValueError(f"qty tidak valid: {data['qty']!r}")
    if versi_kode == 1:
        diskon = data["diskon"]  # bug: diskon null memicu TypeError
    else:
        diskon = data.get("diskon") or 0  # perbaikan: null dianggap 0
    total = data["qty"] * data["harga"] - diskon
    return (data["pesanan_id"], data["waktu"], data["pelanggan"],
            data["produk"], data["qty"], data["harga"], total)


def kelas_galat(galat):
    if isinstance(galat, (json.JSONDecodeError, ValueError, KeyError)):
        return "data"
    return "kode"


def proses_batch(daftar_baris, versi_kode):
    sukses = 0
    for baris, payload in enumerate(daftar_baris, start=1):
        try:
            koneksi.execute("INSERT OR REPLACE INTO pesanan VALUES (?, ?, ?, ?, ?, ?, ?)",
                            transform(payload, versi_kode))
            sukses += 1
        except Exception as galat:
            koneksi.execute(
                """INSERT INTO dead_letter (baris, payload, galat, kelas, percobaan)
                   VALUES (?, ?, ?, ?, 1)""",
                (baris, payload, f"{type(galat).__name__}: {galat}", kelas_galat(galat)))
    koneksi.commit()
    return sukses


def retry_dead_letter(versi_kode):
    for dl_id, baris, payload in koneksi.execute(
        "SELECT dl_id, baris, payload FROM dead_letter ORDER BY baris"
    ).fetchall():
        try:
            koneksi.execute("INSERT OR REPLACE INTO pesanan VALUES (?, ?, ?, ?, ?, ?, ?)",
                            transform(payload, versi_kode))
            koneksi.execute("DELETE FROM dead_letter WHERE dl_id = ?", (dl_id,))
            print(f"  dead letter baris {baris}: SUKSES")
        except Exception as galat:
            koneksi.execute(
                "UPDATE dead_letter SET percobaan = percobaan + 1, galat = ? WHERE dl_id = ?",
                (f"{type(galat).__name__}: {galat}", dl_id))
            print(f"  dead letter baris {baris}: GAGAL lagi ({galat})")
    koneksi.commit()


batch_1 = [
    '{"pesanan_id": 1, "waktu": "2026-01-10T09:15:00", "pelanggan": "Budi", "produk": "Kemeja Batik", "qty": 1, "harga": 250000, "diskon": 0}',
    '{"pesanan_id": 2, "waktu": "2026-01-10T14:30:00", "pelanggan": "Siti", "produk": "Kain Songket", "qty": 2, "harga": 180000, "diskon": 20000}',
    '{"pesanan_id": 3, "waktu": "2026-01-10T19:45:00", "pelanggan": "Dewi", ',
    '{"pesanan_id": 4, "waktu": "2026-01-11T08:05:00", "pelanggan": "Rina", "produk": "Seluar Linen", "harga": 210000, "diskon": 0}',
    '{"pesanan_id": 5, "waktu": "2026-01-11T13:20:00", "pelanggan": "Budi", "produk": "Rok Tenun", "qty": 2, "harga": 150000, "diskon": 0}',
    '{"pesanan_id": 6, "waktu": "2026-01-11T16:40:00", "pelanggan": "Siti", "produk": "Kemeja Batik", "qty": -1, "harga": 250000, "diskon": 0}',
    '{"pesanan_id": 7, "waktu": "2026-01-12T07:40:00", "pelanggan": "Dewi", "produk": "Kain Songket", "qty": 2, "harga": 180000, "diskon": 10000}',
    '{"pesanan_id": 8, "waktu": "2026-01-12T10:05:00", "pelanggan": "Budi", "produk": "Blouse Katun", "qty": 1, "harga": 120000, "diskon": null}',
    '{"pesanan_id": 9, "waktu": "2026-01-12T13:30:00", "pelanggan": "Rina", "produk": "Kain Songket", "qty": 2, "harga": 180000, "diskon": 0}',
    '{"pesanan_id": 10, "waktu": "2026-01-12T17:55:00", "pelanggan": "Siti", "produk": "Seluar Linen", "qty": 1, "harga": 210000, "diskon": 0}',
]

print("=== Batch 1: 10 baris, transform kode v1 ===")
sukses = proses_batch(batch_1, versi_kode=1)
print(f"  {sukses} baris sukses, {10 - sukses} baris dikarantina")

print("=== Diagnosis karantina ===")
for baris, galat, kelas in koneksi.execute(
    "SELECT baris, galat, kelas FROM dead_letter ORDER BY baris"
):
    print(f"  baris {baris} [kelas: {kelas}] {galat}")
saran = {"data": "perbaiki di sumber, lalu minta pengiriman ulang",
         "kode": "perbaiki logika transform, lalu jalankan retry"}
for kelas, jumlah in koneksi.execute(
    "SELECT kelas, COUNT(*) FROM dead_letter GROUP BY kelas"
):
    print(f"  kelas {kelas}: {jumlah} baris -> {saran[kelas]}")

print("=== Perbaikan: baris 3 dan 6 dikirim ulang, kode naik ke v2 ===")
for baris, payload_baru in [
    (3, '{"pesanan_id": 3, "waktu": "2026-01-10T19:45:00", "pelanggan": "Dewi", "produk": "Blouse Katun", "qty": 1, "harga": 120000, "diskon": 0}'),
    (6, '{"pesanan_id": 6, "waktu": "2026-01-11T16:40:00", "pelanggan": "Siti", "produk": "Kemeja Batik", "qty": 1, "harga": 250000, "diskon": 0}'),
]:
    koneksi.execute("UPDATE dead_letter SET payload = ? WHERE baris = ?", (payload_baru, baris))
koneksi.commit()

print("=== Retry per baris (kode v2) ===")
retry_dead_letter(versi_kode=2)

print("=== Hasil akhir ===")
(jumlah_pesanan,) = koneksi.execute("SELECT COUNT(*) FROM pesanan").fetchone()
(jumlah_dl,) = koneksi.execute("SELECT COUNT(*) FROM dead_letter").fetchone()
print(f"  tabel pesanan: {jumlah_pesanan} baris")
print(f"  dead letter tersisa: {jumlah_dl} baris", end=" ")
for baris, percobaan in koneksi.execute("SELECT baris, percobaan FROM dead_letter"):
    print(f"(baris {baris}, percobaan {percobaan}) -> menunggu perbaikan sumber")

koneksi.close()

Output yang Dihasilkan:

text
=== Batch 1: 10 baris, transform kode v1 ===
  6 baris sukses, 4 baris dikarantina
=== Diagnosis karantina ===
  baris 3 [kelas: data] JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 72 (char 71)
  baris 4 [kelas: data] ValueError: kolom wajib hilang: qty
  baris 6 [kelas: data] ValueError: qty tidak valid: -1
  baris 8 [kelas: kode] TypeError: unsupported operand type(s) for -: 'int' and 'NoneType'
  kelas data: 3 baris -> perbaiki di sumber, lalu minta pengiriman ulang
  kelas kode: 1 baris -> perbaiki logika transform, lalu jalankan retry
=== Perbaikan: baris 3 dan 6 dikirim ulang, kode naik ke v2 ===
=== Retry per baris (kode v2) ===
  dead letter baris 3: SUKSES
  dead letter baris 4: GAGAL lagi (kolom wajib hilang: qty)
  dead letter baris 6: SUKSES
  dead letter baris 8: SUKSES
=== Hasil akhir ===
  tabel pesanan: 9 baris
  dead letter tersisa: 1 baris (baris 4, percobaan 2) -> menunggu perbaikan sumber

Mari kita telaah baris-baris kode di atas:

Baris 33-46 (transform): parsing, validasi kolom wajib, dan validasi qty berjalan berurutan sebelum perhitungan total; parameter versi_kode menghidupkan dua perilaku — v1 mengakses data["diskon"] secara langsung (bug terhadap null), v2 mengganti null dengan 0.
Baris 49-52 (kelas_galat): heuristik klasifikasi — galat parsing, validasi, dan kolom hilang dianggap cacat data; galat tipe seperti TypeError dianggap indikasi bug kode.
Baris 55-68 (proses_batch): setiap baris dibungkus try/except; baris sehat masuk tabel pesanan, baris gagal masuk dead_letter lengkap dengan payload mentah dan pesan galat, sehingga batch tidak pernah dibatalkan.
Baris 71-85 (retry_dead_letter): setiap baris karantina diuji ulang satu per satu — sukses dipindahkan ke pesanan lalu dihapus dari karantina, gagal menambah kolom percobaan dan tetap menunggu.

Alur hasilnya persis seperti yang diharapkan: enam baris langsung sehat, empat dikarantina, tiga terklasifikasi cacat data dan satu bug kode. Setelah baris 3 dan 6 dikirim ulang dalam bentuk benar dan transform naik ke v2, retry memindahkan tiga baris lagi ke tabel pesanan; hanya baris 4 yang tetap gagal karena kolom qty memang tidak ada di sumbernya — tidak ada kode yang bisa menyelamatkan data yang tidak pernah dikirim.

Dua catatan menjaga pola tersebut tetap sehat. Pertama, klasifikasi kelas di baris 49-52 hanyalah heuristik: TypeError bisa juga berarti data bertipe salah, dan KeyError bisa berarti transform mengakses kolom yang memang seharusnya opsional — periksa payload mentah sebelum memutuskan siapa yang harus memperbaiki. Kedua, dead letter menuntut konteks yang cocok: untuk batch yang harus atomik (misalnya posting jurnal keuangan yang setengah terproses justru berbahaya), pola yang tepat tetap transaksi penuh all-or-nothing, bukan karantina per baris.

Pola dead letter menyelamatkan baris per baris. Namun ketika bug ditemukan pada logika transform itu sendiri, seluruh hasil sebelumnya — termasuk yang tampak sukses — patut dicurigai dan perlu diproses ulang. Chapter Pemrosesan Ulang Data membahas cara melakukannya dengan aman.