Pipeline terjadwal pasti dijalankan lebih dari sekali untuk tanggal yang sama: alarm yang menyala dua kali, operator yang belum yakin, atau proses yang berhenti di tengah jalan lalu diulang. Pipeline yang baik menyambut eksekusi ulang dengan dua sikap: melewatinya bila hasilnya sudah ada, dan menghasilkan hasil yang sama persis bila dipaksa mengulang. Kita mencapai keduanya dengan satu berkas state JSON sebagai watermark per tanggal, ditambah semantik upsert yang sudah dibangun pada Bagian 4.

Watermark di state.json #

data/state/state.json mencatat satu entri per tanggal partisi. Berikut isi sebenarnya setelah tiga partisi dan satu eksekusi ulang:

json data/state/state.json (entri 2026-11-02, asli) 
{
  "pipeline": "etl-pesanan-tokokita",
  "runs": {
    "2026-11-02": {
      "attempt": 2,
      "status": "success",
      "updated_at": "2026-10-17T07:50:17+00:00",
      "rows": {
        "extracted": 8, "karantina": 0, "duplikat": 0,
        "fact": 8, "summary": 4
      },
      "checksums": {
        "fact_2026-11-02": "597d2f68aa8be0d00a0da7b14aab616c1f30cacdeae849422772dde270318ec1",
        "summary_2026-11-02": "59da51bb74631b00653ff899f1df1012d934b4152875c981335774bb28122333"
      },
      "gold": "data/gold/summary_harian_2026-11-02.csv"
    }
  }
}

Entri menyimpan status, jumlah percobaan (attempt), jumlah baris, dan checksum hasil. Checksum inilah yang nanti dibandingkan untuk membuktikan bahwa eksekusi ulang tidak mengubah apa pun. Kelas StateStore mengelola berkas ini:

python src/idempotency.py — inti StateStore (dipangkas) 
def save(self) -> None:
    self.path.parent.mkdir(parents=True, exist_ok=True)
    sementara = self.path.with_suffix(".json.tmp")
    sementara.write_text(
        json.dumps(self.data, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )
    os.replace(sementara, self.path)

def has_success(self, tanggal) -> bool:
    return self._run(str(tanggal)).get("status") == "success"

def _catat(self, tanggal, status: str, muatan: dict) -> None:
    kunci = str(tanggal)
    run = self.data["runs"].setdefault(kunci, {})
    run["attempt"] = int(run.get("attempt", 0)) + 1
    run["status"] = status
    for k, v in muatan.items():
        if k != "attempt":
            run[k] = v

Dua keputusan di sini layak diperhatikan:

  1. save() menulis ke berkas .tmp lalu os.replace: os.replace mengganti berkas secara atomik pada level sistem berkas. Proses yang berhenti di tengah penulisan tidak meninggalkan state.json setengah jadi; hasil terburuk adalah berkas .tmp yang tak terpakai.
  2. attempt bertambah pada setiap pencatatan: keberhasilan maupun kegagalan sama-sama dihitung. Jumlah percobaan inilah yang dipakai scheduler pada Bagian 6 untuk memblokir tanggal yang terus-menerus gagal.

Jalur Lewat dan --force #

Bagian penentu keputusan berada di awal run_for_date:

python src/pipeline.py — jalur lewat (dipangkas) 
if state.has_success(order_date) and not force:
    lama = state.get(order_date)
    m = akhir("skipped", lama.get("rows", {}), lama.get("checksums", {}),
              lama.get("gold", ""), None, state.attempts(order_date))
    write_manifest(m, runs_dir)
    print(f"[lewati] {kunci} sudah sukses pada percobaan sebelumnya "
          "(pakai --force untuk memproses ulang)")
    ...
    return m

Eksekusi kedua tanpa --force berhenti di sini:

Output yang Dihasilkan:

text 
[lewati] 2026-11-02 sudah sukses pada percobaan sebelumnya (pakai --force untuk memproses ulang)
[selesai] run-2026-11-02-075015 status=skipped durasi=0ms exit=0

Status tetap exit=0: melewati pekerjaan yang sudah selesai bukanlah kegagalan. Manifest tetap ditulis dengan status skipped sehingga riwayat eksekusi lengkap.

Bukti Eksekusi Ulang Identik #

Sekarang kita paksa mengulang tanggal yang sama dengan --force, lalu bandingkan ketiga manifestnya:

bash 
python3 -m src.pipeline --date 2026-11-02 --force

Output yang Dihasilkan:

text 
[extract] pesanan_2026-11-02.csv: 8 baris, 771 bytes
[transform] 8 baris -> 8 fact, 0 karantina, 0 duplikat
[load] MERGE fact: 8 baris; summary 2026-11-02: 4 baris
[validate] OK  reconcile kota: 8 baris di 4 kota (total sumber 8 = total target 8)
[validate] OK  agregat harian: fact: 8 baris, Rp3,542,000.00 bruto, Rp3,092,000.00 bersih; summary: 8 baris, Rp3,542,000.00 bruto, Rp3,092,000.00 bersih
[validate] OK  gold csv: 4 baris gold vs 4 baris summary
[gold] data/gold/summary_harian_2026-11-02.csv
[selesai] run-2026-11-02-075017 status=success durasi=46ms exit=0

Output yang Dihasilkan:

text 
== manifest eksekusi 2026-11-02 (urut waktu) ==
run-2026-11-02-075011: status=success attempt=1 rows=8 durasi=43ms
  checksum fact   : 597d2f68aa8be0d00a0da7b14aab616c1f30cacdeae849422772dde270318ec1
  checksum summary: 59da51bb74631b00653ff899f1df1012d934b4152875c981335774bb28122333
run-2026-11-02-075015: status=skipped attempt=1 rows=8 durasi=0ms
  checksum fact   : 597d2f68aa8be0d00a0da7b14aab616c1f30cacdeae849422772dde270318ec1
  checksum summary: 59da51bb74631b00653ff899f1df1012d934b4152875c981335774bb28122333
run-2026-11-02-075017: status=success attempt=2 rows=8 durasi=46ms
  checksum fact   : 597d2f68aa8be0d00a0da7b14aab616c1f30cacdeae849422772dde270318ec1
  checksum summary: 59da51bb74631b00653ff899f1df1012d934b4152875c981335774bb28122333

Ketiga checksum fact dan checksum summary bernilai sama. Eksekusi ulang menghasilkan warehouse yang identik di tingkat baris, bukan sekadar jumlah baris yang kebetulan sama. Pengujian otomatisnya juga ada: test_rerun_force_identik menjalankan skenario yang sama pada direktori uji dan membandingkan checksum dari state.

Kombinasi ini adalah wujud dari chapter Idempotent Transformation dan Idempotent Ingestion: upsert kunci untuk fact, ganti partisi untuk summary, dan watermark untuk mengetahui pekerjaan mana yang belum selesai. Watermark juga berfungsi sebagai checkpoint murah seperti dibahas chapter Checkpointing, dan memungkinkan pemrosesan ulang yang dibahas Pemrosesan Ulang Data. Pada bagian berikutnya kita menyerahkan keputusan “tanggal apa yang harus diproses” kepada penjadwal.

Bacaan lanjut: os.replace pada dokumentasi Python.