Semua mekanisme keandalan yang kita bangun sejauh ini punya efek samping yang sama: pengulangan. Semantik at-least-once pada Chapter Ingestion dari Message Queue mengirim ulang pesan yang gagal diproses, dan retry pada Chapter Rate Limiting dan Retries mengirim ulang permintaan yang ditolak. Dalam pipeline nyata, pesanan yang sama bisa tiba dua atau tiga kali di warehouse — dan jika pipeline kita menyalin apa adanya, laporan penjualan menjadi menggelembung dua kali lipat. Idempotency adalah sifat yang menyelamatkan kita: pipeline dijalankan dua kali, hasilnya tetap sama. Chapter ini membahas cara mendesain idempotency lewat dedup key, pola upsert, dan file staging, lalu membuktikannya dengan ingestion yang dijalankan dua kali.

Apa Itu Idempotent #

Sebuah operasi disebut idempotent jika menjalankannya sekali atau berkali-kali menghasilkan keadaan akhir yang sama. Contoh paling mudah ada di SQL: DELETE FROM pesanan WHERE order_id = 'ORD-001' tetap menghasilkan satu baris terhapus berapa kali pun dijalankan, karena eksekusi kedua tidak lagi menemukan baris tersebut. Sebaliknya, INSERT INTO laporan VALUES (...) tanpa batasan keunikan menambah baris baru setiap eksekusi — operasi seperti inilah yang membuat pipeline gagal di produksi.

Dalam ingestion, idempotency berarti satu baris data dari sumber hanya boleh menghasilkan satu baris di tujuan, berapa kali pun proses ingestion berjalan. Ada tiga pilar untuk mencapainya:

  1. Dedup key — kolom yang mengidentifikasi baris secara unik, misalnya order_id.
  2. Operasi penulisan yang aman — INSERT OR IGNORE atau upsert, bukan INSERT polos.
  3. Penulisan file yang atomik — file hasil unduhan tidak pernah terlihat setengah jadi.

Dedup Key dan Pola Upsert #

Dedup key adalah kunci keunikan yang kita pilih dari data sumber, dan database tujuan yang menegakkannya lewat constraint PRIMARY KEY atau UNIQUE. Untuk data pesanan, order_id adalah kandidat alami; untuk aliran event, kombinasi order_id plus waktu event sering diperlukan karena satu pesanan bisa punya banyak event (dibuat, dibayar, dikirim).

Setelah dedup key ditetapkan, ada dua pola penulisan yang aman. INSERT OR IGNORE membiarkan baris pertama menang dan mengabaikan semua salinan berikutnya — cocok untuk data snapshot. Upsert ON CONFLICT ... DO UPDATE (yang kita pakai pada Chapter Change Data Capture) menimpa dengan versi terbaru — cocok ketika event berikutnya memuat informasi lebih baru.

Keduanya sama-sama aman diulang; bedanya hanya versi mana yang menang. Untuk tabel yang menampung event append-only seperti log aktivitas, INSERT OR IGNORE hampir selalu menjadi pilihan yang tepat.

File Staging dan Atomic Move #

Sumber data berbentuk berkas menambah satu masalah: unduhan bisa terputus di tengah jalan, meninggalkan berkas JSONL yang hanya berisi separuh data. Jika pipeline lain membaca berkas setengah jadi tersebut, hasilnya baris yang rusak. Pola staging menyelesaikannya dengan tiga langkah: tulis ke berkas sementara (pesanan.jsonl.staging), periksa kelengkapan, lalu pindahkan ke nama final dengan os.replace(). Sistem operasi menjamin os.replace() bersifat atomik — pembaca hanya melihat berkas lama yang utuh atau berkas baru yang utuh, tidak pernah keadaan di antaranya. Beri juga nama unik pada berkas staging, misalnya akhiran .staging atau id proses, agar dua worker yang berjalan bersamaan tidak saling menimpa berkas staging milik satu sama lain. Rincian jaminan ini tercantum pada dokumentasi os.replace() di https://docs.python.org/3/library/os.html#os.replace .

Contoh: Ingestion Idempotent dengan SQLite #

Kita akan membuktikan idempotency secara langsung. Sumber berupa tujuh baris pesanan yang mengandung dua duplikat (ORD-002 dan ORD-004 masing-masing tampil dua kali), meniru pengiriman ulang dari message queue. Fungsi ingestion yang sama kemudian dijalankan dua kali; jika desain kita benar, jumlah baris di tabel tidak bertambah pada eksekusi kedua.

python 
import json
import os
import sqlite3
import tempfile

BARIS_SUMBER = [
    {"order_id": "ORD-001", "pelanggan": "Budi", "total": 150000},
    {"order_id": "ORD-002", "pelanggan": "Siti", "total": 275000},
    {"order_id": "ORD-003", "pelanggan": "Dewi", "total": 99000},
    {"order_id": "ORD-002", "pelanggan": "Siti", "total": 275000},
    {"order_id": "ORD-004", "pelanggan": "Budi", "total": 430000},
    {"order_id": "ORD-005", "pelanggan": "Dewi", "total": 87000},
    {"order_id": "ORD-004", "pelanggan": "Budi", "total": 430000},
]

direktori = tempfile.mkdtemp(prefix="ingest-")
berkas_final = os.path.join(direktori, "pesanan.jsonl")
berkas_staging = berkas_final + ".staging"


def unduh_ke_staging():
    with open(berkas_staging, "w", encoding="utf-8") as f:
        for baris in BARIS_SUMBER:
            f.write(json.dumps(baris) + "\n")
    os.replace(berkas_staging, berkas_final)  # pemindahan atomik


def jalankan_ingestion(db):
    db.execute("""CREATE TABLE IF NOT EXISTS pesanan (
        order_id TEXT PRIMARY KEY,
        pelanggan TEXT NOT NULL,
        total INTEGER NOT NULL)""")
    with open(berkas_final, encoding="utf-8") as f:
        baris_data = [json.loads(baris) for baris in f]
    masuk = 0
    for baris in baris_data:
        kursor = db.execute(
            "INSERT OR IGNORE INTO pesanan (order_id, pelanggan, total) VALUES (?, ?, ?)",
            (baris["order_id"], baris["pelanggan"], baris["total"]),
        )
        masuk += kursor.rowcount
    db.commit()
    total = db.execute("SELECT COUNT(*) FROM pesanan").fetchone()[0]
    print(f"baris dibaca: {len(baris_data)}, baru dimasukkan: {masuk}, "
          f"duplikat diabaikan: {len(baris_data) - masuk}, total tabel: {total}")


db = sqlite3.connect(":memory:")
unduh_ke_staging()
print("jalankan 1 (proses berjalan normal):")
jalankan_ingestion(db)
print("jalankan 2 (proses tanpa sengaja berjalan lagi):")
jalankan_ingestion(db)

Output yang Dihasilkan:

text 
jalankan 1 (proses berjalan normal):
baris dibaca: 7, baru dimasukkan: 5, duplikat diabaikan: 2, total tabel: 5
jalankan 2 (proses tanpa sengaja berjalan lagi):
baris dibaca: 7, baru dimasukkan: 0, duplikat diabaikan: 7, total tabel: 5

Mari kita telaah baris-baris kode di atas:

  1. Baris 21-25 (unduh_ke_staging): seluruh baris ditulis ke berkas_staging lebih dulu; hanya setelah berkas tertutup rapi, os.replace() memindahkannya ke berkas_final dalam satu langkah atomik.
  2. Baris 29-32 (CREATE TABLE IF NOT EXISTS): kolom order_id ditetapkan sebagai PRIMARY KEY — inilah dedup key yang ditegakkan oleh SQLite, bukan oleh pengecekan di kode Python.
  3. Baris 37-40 (INSERT OR IGNORE): saat order_id sudah ada di tabel, SQLite mengabaikan penulisan tersebut tanpa memunculkan galat, sehingga pipeline tidak perlu berhenti karena duplikat.
  4. Baris 41 (masuk += kursor.rowcount): rowcount bernilai 1 saat baris benar-benar masuk dan 0 saat diabaikan, sehingga laporan bisa membedakan baris baru dari duplikat.
  5. Baris 50-53: fungsi jalankan_ingestion yang persis sama dipanggil dua kali — tidak ada kode khusus “mode kedua”, karena idempotency memang bukan trik ekstra melainkan sifat bawaan dari desain penulisannya.

Output di atas adalah buktinya. Eksekusi pertama memasukkan 5 baris karena 2 duplikat diabaikan. Eksekusi kedua membaca ulang ke-7 baris yang sama, namun INSERT OR IGNORE menolak semuanya: baris baru 0, duplikat diabaikan 7, dan total tabel tetap 5. Seandainya scheduler menjadwalkan ulang pipeline ini sepuluh kali pun, jumlah baris di warehouse tidak akan berubah.

Pola staging ikut berperan pada ketahanan ini: jika proses berhenti di tengah unduhan, hanya berkas staging yang tertinggal setengah jadi, sementara berkas final tetap utuh dan siap dipakai eksekusi berikutnya.

Idempotency memastikan tidak ada baris ganda di warehouse, tetapi tidak memeriksa isi barisnya: total bernilai negatif, nama pelanggan kosong, atau harga berupa teks tetap lolos masuk. Gerbang pemeriksaan terakhir dibangun pada Chapter Validasi Data saat Ingestion.