Pipeline Recovery

Pukul 02:00 pipeline pesanan harian toko online kita berhenti di partisi terakhir: satu baris JSON dari sistem kasir terpotong, dan proses batal di tengah jalan. Laporan penjualan harus tayang pukul 06:00, sementara mengulang seluruh proses dari nol memakan waktu dua jam. Strategi pemulihan menjawab dilema tersebut: pipeline mengetahui persis bagian mana yang sudah selesai, memvalidasi bagian yang setengah jadi, lalu hanya memproses ulang yang bermasalah — dan semua pihak di hilir tahu bahwa run semalam gagal.

Tiga Strategi Pemulihan #

Kegagalan pipeline punya beberapa wajah, dan setiap wajah menuntut strategi berbeda. Retry task mengulang task yang sama beberapa kali dengan jeda — tepat untuk kegagalan transien seperti jaringan yang terputus sejenak atau sumber yang sedang sibuk, dengan syarat task bersangkutan idempotent seperti yang dibahas pada Chapter Idempotency dan Retries. Resume dari checkpoint melanjutkan run dari titik terakhir yang tercatat tuntas, mengikuti pola state yang dibangun pada Chapter Checkpointing. Re-run partisi memproses ulang satu partisi setelah akar masalahnya diperbaiki, misalnya setelah sumber mengirim ulang berkas yang rusak.

Strategi	Kapan dipakai	Syarat aman
Retry task	kegagalan transien: jaringan, sumber sibuk sesaat	task idempotent
Resume dari checkpoint	run panjang berhenti di tengah, sebagian stage tuntas	checkpoint ditulis hanya setelah stage benar-benar selesai
Re-run partisi	data sumber rusak lalu diperbaiki, atau bug kode diperbaiki	muatan idempotent, misalnya `INSERT OR REPLACE`

Data Corrupted dan Komunikasi ke Downstream #

Baris yang korup tidak sembuh dengan retry — JSON yang terpotong akan tetap terpotong pada percobaan kedua. Penanganan yang tepat adalah mengkarantina baris bermasalah atau meminta sumber mengirim ulang, seperti dibahas pada Chapter Penanganan Data yang Gagal, lalu menjalankan re-run untuk partisi tersebut. Sementara itu, job downstream — dashboard harian, agregat mingguan — wajib tahu bahwa run semalam gagal agar tidak menerbitkan laporan dari data setengah jadi. Pola yang lazim: pipeline menulis status setiap run ke tabel kecil, dan downstream memeriksa status tersebut sebelum membaca tabel hasil. Orkestrator seperti Apache Airflow menyediakan status per task secara bawaan; rujukannya ada di dokumentasi Apache Airflow .

Implementasi: Recovery Run dengan Validasi Output Parsial #

Skrip berikut memodelkan pipeline harian tiga partisi. Koneksi SQLite sengaja memakai mode autocommit agar setiap baris langsung permanen — meniru pipeline produksi yang bisa berhenti kapan saja dan meninggalkan hasil parsial. Pada run pertama, partisi ketiga gagal karena baris ke-9 korup; setelah sumber mengirim ulang baris tersebut, run kedua memvalidasi tiap partisi dan hanya memproses ulang yang perlu.

python
"""Pipeline harian toko online: demo pemulihan dari kegagalan di tengah partisi."""
import json
import sqlite3
from pathlib import Path

DATA = Path("data")
DB = Path("toko_pulih.db")
if DB.exists():
    DB.unlink()

PARTISI = ["2026-01-10", "2026-01-11", "2026-01-12"]

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

DATA.mkdir(exist_ok=True)
for nama, baris_baris in Sumber.items():
    (DATA / f"partisi_{nama}.jsonl").write_text("\n".join(baris_baris) + "\n")

koneksi = sqlite3.connect(DB, isolation_level=None)  # autocommit: baris langsung tersimpan
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 checkpoint_partisi (
    partisi TEXT PRIMARY KEY, status TEXT NOT NULL, jumlah_baris INTEGER NOT NULL);
CREATE TABLE status_run (
    run_id INTEGER PRIMARY KEY, status TEXT NOT NULL,
    partisi_terpengaruh TEXT NOT NULL, pesan TEXT NOT NULL);
""")


def baca_partisi(partisi):
    return (DATA / f"partisi_{partisi}.jsonl").read_text().splitlines()


def proses_partisi(partisi):
    """Muat satu partisi; tulis checkpoint hanya setelah partisi utuh masuk."""
    koneksi.execute(
        "INSERT OR REPLACE INTO checkpoint_partisi VALUES (?, 'proses', 0)", (partisi,))
    jumlah = 0
    for baris in baca_partisi(partisi):
        data = json.loads(baris)
        total = data["qty"] * data["harga"] - (data.get("diskon") or 0)
        koneksi.execute(
            "INSERT OR REPLACE INTO pesanan VALUES (?, ?, ?, ?, ?, ?, ?)",
            (data["pesanan_id"], data["waktu"], data["pelanggan"],
             data["produk"], data["qty"], data["harga"], total))
        jumlah += 1
    koneksi.execute(
        "UPDATE checkpoint_partisi SET status = 'selesai', jumlah_baris = ? WHERE partisi = ?",
        (jumlah, partisi))
    return jumlah


def validasi_partisi(partisi):
    """Bandingkan baris di tabel hasil dengan baris di berkas sumber."""
    (di_db,) = koneksi.execute(
        "SELECT COUNT(*) FROM pesanan WHERE waktu LIKE ?", (f"{partisi}%",)).fetchone()
    di_sumber = len(baca_partisi(partisi))
    return di_db, di_sumber


def jalankan_pipeline(run_id):
    for partisi in PARTISI:
        status = koneksi.execute(
            "SELECT status FROM checkpoint_partisi WHERE partisi = ?", (partisi,)
        ).fetchone()
        if status is None:
            try:
                jumlah = proses_partisi(partisi)
                print(f"  partisi {partisi}: {jumlah} baris dimuat, checkpoint selesai")
            except Exception as galat:
                pesan = f"{type(galat).__name__}: {galat}"
                koneksi.execute(
                    "INSERT OR REPLACE INTO status_run VALUES (?, 'gagal', ?, ?)",
                    (run_id, partisi, pesan))
                print(f"  RUN {run_id} GAGAL di partisi {partisi}: {pesan}")
                return False
        else:
            di_db, di_sumber = validasi_partisi(partisi)
            if status[0] == "selesai" and di_db == di_sumber:
                print(f"  partisi {partisi}: checkpoint selesai, validasi {di_db}/{di_sumber} -> lewati")
            else:
                print(f"  partisi {partisi}: {status[0]}, validasi {di_db}/{di_sumber} -> parsial, proses ulang")
                jumlah = proses_partisi(partisi)
                print(f"  partisi {partisi}: {jumlah} baris dimuat, checkpoint selesai")
    koneksi.execute(
        "INSERT OR REPLACE INTO status_run VALUES (?, 'sukses', '-', 'semua partisi selesai')",
        (run_id,))
    print(f"  RUN {run_id} SUKSES: {len(PARTISI)}/{len(PARTISI)} partisi selesai")
    return True


print("=== Run 1 (malam ini, pukul 02:00) ===")
jalankan_pipeline(1)

print("=== Sumber diperbaiki: baris 9 dikirim ulang lengkap ===")
baris = Sumber["2026-01-12"]
baris[2] = '{"pesanan_id": 9, "waktu": "2026-01-12T14:10:00", "pelanggan": "Rina", "produk": "Seluar Linen", "qty": 1, "harga": 210000, "diskon": 0}'
(DATA / "partisi_2026-01-12.jsonl").write_text("\n".join(baris) + "\n")

print("=== Run 2 (pemulihan, pukul 05:30) ===")
jalankan_pipeline(2)

print("=== Status akhir (yang dibaca job downstream) ===")
for run_id, status, partisi, pesan in koneksi.execute(
    "SELECT run_id, status, partisi_terpengaruh, pesan FROM status_run ORDER BY run_id"
):
    print(f"  run {run_id}: {status} (partisi: {partisi}) {pesan}")
(total,) = koneksi.execute("SELECT COUNT(*) FROM pesanan").fetchone()
print(f"  tabel pesanan: {total} baris")
for partisi in PARTISI:
    (jumlah,) = koneksi.execute(
        "SELECT COUNT(*) FROM pesanan WHERE waktu LIKE ?", (f"{partisi}%",)).fetchone()
    print(f"  partisi {partisi}: {jumlah} baris")

koneksi.close()

Output yang Dihasilkan:

text
=== Run 1 (malam ini, pukul 02:00) ===
  partisi 2026-01-10: 3 baris dimuat, checkpoint selesai
  partisi 2026-01-11: 3 baris dimuat, checkpoint selesai
  RUN 1 GAGAL di partisi 2026-01-12: JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 72 (char 71)
=== Sumber diperbaiki: baris 9 dikirim ulang lengkap ===
=== Run 2 (pemulihan, pukul 05:30) ===
  partisi 2026-01-10: checkpoint selesai, validasi 3/3 -> lewati
  partisi 2026-01-11: checkpoint selesai, validasi 3/3 -> lewati
  partisi 2026-01-12: proses, validasi 2/4 -> parsial, proses ulang
  partisi 2026-01-12: 4 baris dimuat, checkpoint selesai
  RUN 2 SUKSES: 3/3 partisi selesai
=== Status akhir (yang dibaca job downstream) ===
  run 1: gagal (partisi: 2026-01-12) JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 72 (char 71)
  run 2: sukses (partisi: -) semua partisi selesai
  tabel pesanan: 10 baris
  partisi 2026-01-10: 3 baris
  partisi 2026-01-11: 3 baris
  partisi 2026-01-12: 4 baris

Mari kita telaah baris-baris kode di atas:

Baris 36 (isolation_level=None): koneksi berjalan di mode autocommit, sehingga setiap INSERT langsung permanen. Run yang berhenti mendadak meninggalkan baris yang sudah masuk — persis kondisi yang harus dihadapi proses pemulihan.
Baris 38-46 (tiga tabel): pesanan menampung hasil, checkpoint_partisi mencatat status per partisi, dan status_run menjadi kanal komunikasi ke downstream.
Baris 54-70 (proses_partisi): checkpoint berstatus proses ditulis sebelum memuat, dan status selesai hanya ditulis setelah seluruh baris partisi masuk. Kegagalan di tengah meninggalkan status proses — sinyal bahwa partisi tersebut patut dicurigai.
Baris 73-78 (validasi_partisi): menghitung baris di tabel hasil untuk satu partisi lalu membandingkannya dengan jumlah baris di berkas sumber.
Baris 86-96 (partisi baru): partisi tanpa checkpoint diproses; galat apa pun ditangkap, dicatat ke status_run dengan pesan lengkap, lalu run berhenti tanpa melanjutkan partisi berikutnya.
Baris 97-104 (partisi ber-checkpoint): checkpoint selesai yang lolos validasi dilewati; checkpoint proses atau hitungan yang tidak cocok dianggap parsial dan diproses ulang — aman karena muatan memakai INSERT OR REPLACE.

Hasil run kedua menunjukkan pemulihan yang hemat: dua partisi sehat cukup divalidasi lalu dilewati, dan hanya partisi 12 yang diproses ulang menjadi 4 baris utuh. Tabel status_run menyimpan jejak kedua run, sehingga dashboard tahu run 1 gagal di partisi mana dan run 2 sudah menuntaskan pekerjaan.

Pemulihan mengembalikan pipeline ke kondisi layak layanan dalam hitungan menit. Namun setelah insiden biasanya menyusul pertanyaan dari tim bisnis: data yang mana hasil kode versi lama, dan yang mana hasil versi baru? Chapter Pipeline Versioning membahas cara mencatat versi kode dan versi skema output agar setiap angka bisa ditelusuri asal-usulnya.