Checkpointing

Pipeline agregat mingguan TokoKita memproses lima berkas partisi dan berjalan cukup lama. Ketika galat sementara muncul di partisi ketiga, seluruh lima partisi diulang dari awal karena pipeline tidak mengingat progresnya. Masalah ini tidak diselesaikan oleh idempotency: eksekusi ulang memang aman, tetapi tetap membayar penuh biaya komputasi untuk pekerjaan yang sudah selesai. Checkpointing menyelesaikannya dengan menyimpan progres di tengah eksekusi, sehingga saat pipeline dijalankan ulang, eksekusi baru melanjutkan dari titik terakhir, bukan dari baris pertama.

Menyimpan Progres di Tengah Pipeline #

Inti checkpointing adalah memilih unit kerja yang bisa ditandai selesai. Setiap kali satu unit selesai diproses, pipeline mencatat unit tersebut ke berkas state, misalnya out/checkpoint.json, yang berisi daftar partisi selesai beserta hasil agregatnya. Pada eksekusi berikutnya, pipeline memuat state tersebut lebih dulu, melewatkan semua partisi yang sudah tercatat, dan hanya memproses sisanya.

Mekanisme ini hanya berguna bila digabungkan dengan disiplin dari Chapter Idempotency dan Retries. Checkpoint membuat resume murah; idempotency membuat resume aman. Satu tanpa yang lain tetap meninggalkan lubang: resume murah yang tidak aman merusak data, sementara resume aman yang tidak murah tetap membayar komputasi penuh.

Granularitas Checkpoint #

Granularitas menentukan seberapa halus progres dicatat. Tiga pilihan umum disusun dari paling halus ke paling kasar berikut ini.

Trade-off Menulis State #

Checkpoint bukan gratis. Setiap penulisan state adalah pekerjaan ekstra: marshaling ke JSON, menulis ke disk, dan kadang memvalidasi kembali saat dibaca. Granularitas per baris membuat biaya tersebut muncul jutaan kali sehingga bisa melampaui biaya pemrosesan data itu sendiri. Granularitas per tahap nyaris gratis, tetapi satu kegagalan membatalkan seluruh tahap. Aturan praktisnya: pilih unit yang kira-kira sama dengan biaya satu kegagalan yang sanggup ditoleransi. Untuk pipeline agregat dengan partisi harian, partisi adalah unit yang tepat, seperti contoh kita berikut ini.

Contoh: Gagal di Tengah, Resume Menyelesaikan #

Sumber adalah lima berkas di data/orders, yaitu partisi 15 sampai 19 Januari. Pipeline mengagregasi tiap partisi menjadi jumlah pesanan valid dan total pendapatan, mencatat progresnya ke out/checkpoint.json, dan menulis hasil akhir ke out/agregat_harian.json hanya setelah kelima partisi selesai. Tahap pertama sengaja dijalankan dengan argumen --fail-on 2026-01-17 yang mensimulasikan galat sementara di partisi ketiga.

python3 pipeline_checkpoint.py --fail-on 2026-01-17

Output yang Dihasilkan:

selesai 2026-01-15: 3 valid, Rp1375000
selesai 2026-01-16: 3 valid, Rp1120000
gagal memproses 2026-01-17: galat sementara saat membaca berkas sumber (simulasi)

Eksekusi tersebut berhenti dengan kode keluar 1 setelah dua partisi selesai. Isi out/checkpoint.json pada titik itu memperlihatkan progres yang selamat:

{
  "agregat": {
    "2026-01-15": {
      "jumlah": 3,
      "total": 1375000
    },
    "2026-01-16": {
      "jumlah": 3,
      "total": 1120000
    }
  },
  "selesai": [
    "2026-01-15",
    "2026-01-16"
  ]
}

Tahap kedua dijalankan tanpa argumen, sehingga tidak ada lagi galat yang disimulasikan. Pipeline memuat checkpoint, melewatkan dua partisi yang sudah tercatat, lalu menyelesaikan sisanya.

python3 pipeline_checkpoint.py

Output yang Dihasilkan:

lewati 2026-01-15: sudah ada di checkpoint
lewati 2026-01-16: sudah ada di checkpoint
selesai 2026-01-17: 3 valid, Rp1340000
selesai 2026-01-18: 2 valid, Rp905000
selesai 2026-01-19: 2 valid, Rp1225000
pipeline tuntas: 5 partisi -> out/agregat_harian.json

Skrip lengkapnya adalah sebagai berikut:

"""Pipeline dengan checkpoint per partisi: resume setelah kegagalan."""
import csv
import json
import sys
from pathlib import Path

SUMBER = Path("data/orders")
CHECKPOINT = Path("out/checkpoint.json")
HASIL = Path("out/agregat_harian.json")


def muat_checkpoint():
    if CHECKPOINT.exists():
        return json.loads(CHECKPOINT.read_text())
    return {"selesai": [], "agregat": {}}


def simpan_checkpoint(state):
    CHECKPOINT.parent.mkdir(parents=True, exist_ok=True)
    CHECKPOINT.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n")


def proses_partisi(tanggal, gagal_di):
    if tanggal == gagal_di:
        raise RuntimeError("galat sementara saat membaca berkas sumber (simulasi)")
    with (SUMBER / f"orders_{tanggal}.csv").open() as f:
        baris = list(csv.DictReader(f))
    valid = [b for b in baris if b["status"] != "batal"]
    return {"jumlah": len(valid), "total": sum(int(b["total"]) for b in valid)}


def main():
    gagal_di = None
    if "--fail-on" in sys.argv:
        gagal_di = sys.argv[sys.argv.index("--fail-on") + 1]
    state = muat_checkpoint()
    partisi = sorted(p.stem.removeprefix("orders_") for p in SUMBER.glob("orders_*.csv"))
    for tanggal in partisi:
        if tanggal in state["selesai"]:
            print(f"lewati {tanggal}: sudah ada di checkpoint")
            continue
        try:
            hasil = proses_partisi(tanggal, gagal_di)
        except RuntimeError as galat:
            print(f"gagal memproses {tanggal}: {galat}")
            sys.exit(1)
        state["selesai"].append(tanggal)
        state["agregat"][tanggal] = hasil
        simpan_checkpoint(state)
        print(f"selesai {tanggal}: {hasil['jumlah']} valid, Rp{hasil['total']}")
    HASIL.parent.mkdir(parents=True, exist_ok=True)
    HASIL.write_text(json.dumps(state["agregat"], indent=2, sort_keys=True) + "\n")
    print(f"pipeline tuntas: {len(partisi)} partisi -> {HASIL}")


if __name__ == "__main__":
    main()

Mari kita telaah baris-per-baris kode di atas:

1. Baris 12-16 (muat_checkpoint): fungsi mengembalikan isi checkpoint.json bila berkas ada; pada eksekusi pertama, berkas belum ada sehingga state dimulai kosong.
2. Baris 23-30 (proses_partisi): satu partisi dibaca, pesanan batal dibuang, lalu hasil agregat partisi dikembalikan sebagai dictionary kecil; inilah unit kerja yang menjadi ukuran checkpoint kita.
3. Baris 38-41 (lewat partisi selesai): partisi yang tercatat di state["selesai"] langsung dilewati, sehingga pekerjaan lama tidak dibayar dua kali.
4. Baris 42-45 (jalur kegagalan): galat sementara mencetak laporan lalu keluar dengan kode 1; progres dua partisi sebelumnya tetap tersimpan karena tiap keberhasilan langsung dicatat.
5. Baris 46-50 (catat segera): hasil partisi ditambahkan ke state dan ditulis ke disk pada saat itu juga, bukan ditumpuk di akhir, agar kegagalan berikutnya tidak menghapus progres.

Gabungan dua keluaran di atas adalah bukti kerja checkpointing: eksekusi pertama menyelesaikan dua partisi lalu berhenti di partisi ketiga, dan eksekusi kedua melanjutkan tanpa memproses ulang partisi 15 dan 16, terlihat dari dua baris lewati pada awal keluaran. Hasil akhir out/agregat_harian.json memuat kelima tanggal, konsisten dengan ringkasan per tanggal 15 sampai 19 Januari.

Pipeline kita kini aman diulang dan murah dilanjutkan. Satu pertanyaan terakhir tetap terbuka sebelum pipeline dipercaya menjalankan data produksi: bagaimana membuktikan bahwa semua perilaku ini benar, bukan sekadar terlihat benar? Jawabannya adalah pengujian pipeline, yang dibahas pada Chapter Pipeline Testing.