Pipeline TokoKita berjalan setiap malam, dan kegagalan bukan kejadian langka: koneksi database terputus tiga detik, berkas sumber belum selesai diunggah, atau node scheduler restart tepat di tengah eksekusi. Kebijakan pertama yang terpikir biasanya adalah retry otomatis, tetapi retry hanya aman bila pipeline bersifat idempotent: dieksekusi sekali atau sepuluh kali, hasil akhirnya tetap sama. Tanpa sifat tersebut, setiap retry menambah duplikat baru ke tabel target dan membuat kerusakan makin sulit diperbaiki. Chapter ini merancang pipeline idempotent dan membuktikannya lewat perbandingan hash antara dua eksekusi.
Apa itu Idempotent Pipeline #
Pipeline disebut idempotent bila eksekusi berulang dengan input yang sama selalu menghasilkan kondisi akhir yang sama. Definisi ini berlaku untuk keseluruhan pipeline, bukan untuk satu tahap saja: bila tahap akhir menimpa tabel dengan benar tetapi tahap tengah menulis berkas bertumpuk, pipeline secara keseluruhan belum idempotent.
Sifat ini dibutuhkan pada tiga situasi nyata. Pertama, retry otomatis setelah galat sementara. Kedua, backfill: menjalankan ulang pipeline untuk tanggal lampau karena aturan transformasi berubah. Ketiga, eksekusi tumpang tindih: scheduler menjalankan ulang run yang kemarin belum tercatat selesai. Ketiganya menuntut kesimpulan yang sama, yaitu eksekusi berulang tidak boleh mengubah hasil.
Tiga Pilar Desain #
Desain idempotent berdiri di atas tiga keputusan konkret. Ketiganya sederhana, dan ketiganya sering dilewatkan.
- Path output tetap per run: path hasil diturunkan dari parameter run, misalnya
out/run_2026-01-09/. Path yang memuat waktu eksekusi, sepertiout/run_{timestamp}/, membuat setiap retry menulis ke lokasi baru, meninggalkan sisa output setengah jadi, dan memaksa konsumen menebak lokasi yang benar. - Dedup input: sumber boleh mengirim baris yang sama berkali-kali, misalnya berkas dikirim ulang saat jaringan pulih. Pipeline memilih satu versi terbaru per kunci sebelum menulis.
- Upsert target: penulisan ke tabel target dilakukan berdasarkan kunci dengan
ON CONFLICT ... DO UPDATE, bukanINSERTpolos, sehingga eksekusi ulang menimpa baris lama alih-alih menambah baris baru.
Contoh: Dua Eksekusi, Hasil Identik #
Sumber kita adalah data/orders_2026-01-09.csv yang sengaja memuat duplikat: pesanan 3001 terkirim tiga kali dengan status yang makin baru, dan versi terbarunya berstatus Selesai. Skrip menguji dua pola sekaligus. Pola append menulis tanpa kunci sebagai anti-contoh; pola idempotent menerapkan dedup, upsert, dan path tetap, lalu hash SHA-256 dari isi tabel target dibandingkan antara eksekusi pertama dan eksekusi kedua.
"""Pipeline idempotent TokoKita: dedup + upsert + path tetap, dua eksekusi identik."""
import csv
import hashlib
import json
import sqlite3
from pathlib import Path
SUMBER = Path("data/orders_2026-01-09.csv")
NAIVE = Path("out/naive.db")
WAREHOUSE = Path("out/warehouse.db")
OUT_RUN = Path("out/run_2026-01-09") # path tetap untuk run 2026-01-09
NAIVE.parent.mkdir(parents=True, exist_ok=True)
def baca_sumber():
with SUMBER.open() as f:
return list(csv.DictReader(f))
def dedup(baris):
terpilih = {}
for b in baris:
kunci = int(b["order_id"])
if kunci not in terpilih or b["updated_at"] > terpilih[kunci]["updated_at"]:
terpilih[kunci] = {
"order_id": kunci,
"pelanggan": b["pelanggan"],
"status": b["status"].lower(),
"total": int(b["total"]),
"updated_at": b["updated_at"],
}
return [terpilih[k] for k in sorted(terpilih)]
def muat_naive():
k = sqlite3.connect(NAIVE)
k.execute("""CREATE TABLE IF NOT EXISTS orders (
order_id INTEGER, pelanggan TEXT, status TEXT, total INTEGER)""")
k.executemany(
"INSERT INTO orders VALUES (?, ?, ?, ?)",
[(int(b["order_id"]), b["pelanggan"], b["status"].lower(), int(b["total"]))
for b in baca_sumber()],
)
k.commit()
print(f" append selesai: tabel berisi "
f"{k.execute('SELECT COUNT(*) FROM orders').fetchone()[0]} baris")
k.close()
def muat_idempotent():
baris = dedup(baca_sumber())
k = sqlite3.connect(WAREHOUSE)
k.execute("""CREATE TABLE IF NOT EXISTS orders (
order_id INTEGER PRIMARY KEY, pelanggan TEXT, status TEXT,
total INTEGER, updated_at TEXT)""")
for b in baris:
k.execute("""
INSERT INTO orders VALUES (?, ?, ?, ?, ?)
ON CONFLICT(order_id) DO UPDATE SET
status = excluded.status, total = excluded.total,
updated_at = excluded.updated_at
""", (b["order_id"], b["pelanggan"], b["status"], b["total"], b["updated_at"]))
k.commit()
OUT_RUN.mkdir(parents=True, exist_ok=True)
(OUT_RUN / "ringkasan.json").write_text(json.dumps(
{"jumlah_pesanan": len(baris),
"total_pendapatan": sum(b["total"] for b in baris)}, indent=2) + "\n")
print(f" upsert selesai: {len(baris)} baris, ringkasan di {OUT_RUN / 'ringkasan.json'}")
k.close()
def hash_target():
k = sqlite3.connect(WAREHOUSE)
dump = "\n".join(
f"{o}|{p}|{s}|{t}|{u}"
for o, p, s, t, u in k.execute(
"SELECT order_id, pelanggan, status, total, updated_at FROM orders "
"ORDER BY order_id"))
k.close()
return hashlib.sha256(dump.encode()).hexdigest()[:16]
print("=== pola append tanpa kunci (anti-contoh) ===")
muat_naive()
muat_naive()
print("=== pola idempotent: dedup + upsert + path tetap ===")
muat_idempotent()
hash_1 = hash_target()
muat_idempotent()
hash_2 = hash_target()
print(f" hash eksekusi 1: {hash_1}")
print(f" hash eksekusi 2: {hash_2}")
print(f" hasil identik: {hash_1 == hash_2}")Output yang Dihasilkan:
=== pola append tanpa kunci (anti-contoh) ===
append selesai: tabel berisi 6 baris
append selesai: tabel berisi 12 baris
=== pola idempotent: dedup + upsert + path tetap ===
upsert selesai: 4 baris, ringkasan di out/run_2026-01-09/ringkasan.json
upsert selesai: 4 baris, ringkasan di out/run_2026-01-09/ringkasan.json
hash eksekusi 1: 1dcaacb4ee62b5ad
hash eksekusi 2: 1dcaacb4ee62b5ad
hasil identik: TrueMari kita telaah baris-per-baris kode di atas:
- Baris 21-33 (
dedup): dictionaryterpilihmenyimpan satu versi perorder_id; baris baru menimpa yang lama hanya bilaupdated_at-nya lebih baru, sehingga versi terbaru selalu menang tanpa peduli berapa kali sumber mengirim ulang. - Baris 57-63 (upsert di
muat_idempotent): setiap baris ditulis melaluiON CONFLICT(order_id) DO UPDATE, jadi eksekusi kedua memperbarui baris yang sudah ada alih-alih menambahnya. - Baris 65-68 (path tetap): ringkasan run selalu ditulis ke
out/run_2026-01-09/ringkasan.json; eksekusi ulang menimpa berkas yang sama, tidak pernah membuatout/run_2026-01-09-v2/. - Baris 73-81 (
hash_target): isi tabel target dibariskan dalam urutanorder_id, lalu diringkas menjadi SHA-256; pembanding ini yang menjatuhkan vonis identik atau tidak.
Keluaran anti-contoh memperlihatkan bahayanya dengan cepat: tabel naik dari 6 baris menjadi 12 baris hanya karena eksekusi diulang sekali, padahal sumber tidak berubah. Sebaliknya, pola idempotent tetap berisi 4 baris setelah dua eksekusi, dan kedua hash sama persis sampai digit terakhir yang tercetak, ditutup baris hasil identik: True.
Safe Retry Semantics #
Retry otomatis baru boleh diaktifkan setelah pipeline terbukti idempotent; urutannya tidak boleh dibalik. Uji paling murah untuk membuktikannya persis seperti contoh di atas: jalankan dua kali, bandingkan hash, dan pastikan hasil identik sebelum menaikkan jumlah percobaan ulang di scheduler. Bila hash berbeda, perbaiki desainnya dulu, bukan menambah jumlah retry, karena menambah retry pada pipeline non-idempotent hanya mempercepat penumpukan duplikat.
Klausa ON CONFLICT ... DO UPDATE terdokumentasi di SQLite UPSERT
, dan kebijakan percobaan ulang pada orkestrasi produksi dibahas di dokumentasi Prefect
. Idempotency membuat eksekusi ulang aman dari sisi hasil, tetapi tidak membuat eksekusi ulang menjadi murah: pipeline yang berjalan lima jam tetap membayar lima jam lagi setiap kali gagal di menit ke-280. Untuk itu kita perlu menyimpan progres di tengah eksekusi, yang dibahas pada Chapter Checkpointing.