Incremental Ingestion

Tabel pesanan toko online kita menambah dua ribu baris per hari, sedangkan historinya sudah menembus sepuluh juta baris. Menyalin seluruh tabel setiap malam berarti memindahkan 10.002.000 baris padahal hanya 2.000 baris yang benar-benar baru — pemborosan yang membesar setiap bulan. Incremental ingestion menjawab masalah ini dengan satu ide sederhana: catat posisi terakhir pembacaan, lalu ambil hanya data setelah posisi tersebut. Penanda posisi itulah yang disebut watermark, dan catatannya wajib bertahan antar run agar tidak hilang.

Watermark: Penanda Posisi Terakhir #

Watermark adalah nilai terakhir yang berhasil diproses dari sebuah kolom pembeda — biasanya updated_at atau id. Alur kerjanya tiga langkah: baca watermark dari penyimpanan state, ambil baris dengan nilai kolom lebih besar dari watermark, lalu perbarui watermark setelah data tersimpan aman di tujuan. Run berikutnya melanjutkan dari posisi baru, sehingga pekerjaan tidak diulang dari nol.

Penyimpanan state bisa berupa file JSON untuk skrip kecil, tabel khusus di data warehouse untuk pipeline serius, atau fitur bawaan orchestrator. Prinsipnya sama: state harus bertahan melebihi umur proses, dan hanya diperbarui setelah data benar-benar tersimpan.

Urutan tersebut juga menentukan perilaku saat run gagal di tengah jalan. Bila proses berhenti setelah data tersimpan tetapi sebelum watermark diperbarui, run ulang membaca ulang jendela yang sama — aman selama penulisan ke tujuan bersifat idempotent. Sebaliknya, watermark yang diperbarui sebelum data tersimpan membuat baris terlewat secara permanen. Karena itu urutan “simpan data dulu, perbarui watermark kemudian” bukan pilihan gaya penulisan kode, melainkan syarat kebenaran pipeline.

Memilih Kolom Watermark #

Untuk tabel pesanan yang statusnya bisa berubah dari baru menjadi dikirim, watermark updated_at adalah pilihan tepat karena perubahan pada baris lama tetap terbaca. Untuk tabel event yang sekali tulis tidak pernah berubah, watermark id lebih aman karena nilainya selalu naik tanpa bergantung pada jam aplikasi.

Implementasi: State Watermark di File JSON #

Kita bangun ingestion incremental dari database SQLite sumber.db yang berisi tabel pesanan dengan 6 baris (kolom id, pelanggan, total, updated_at). Skrip berikut menyalin baris baru ke gudang.db dan menyimpan watermark di watermark.json:

import json
import sqlite3
from pathlib import Path

BERKAS_WATERMARK = Path("watermark.json")
WATERMARK_AWAL = "1970-01-01T00:00:00"


def baca_watermark():
    if BERKAS_WATERMARK.exists():
        return json.loads(BERKAS_WATERMARK.read_text())["updated_at_terakhir"]
    return WATERMARK_AWAL


def simpan_watermark(nilai):
    BERKAS_WATERMARK.write_text(
        json.dumps({"updated_at_terakhir": nilai}, indent=2)
    )


def jalankan():
    watermark = baca_watermark()
    print(f"Watermark awal      : {watermark}")

    sumber = sqlite3.connect("sumber.db")
    tujuan = sqlite3.connect("gudang.db")
    tujuan.execute("""
        CREATE TABLE IF NOT EXISTS pesanan (
            id INTEGER PRIMARY KEY,
            pelanggan TEXT NOT NULL,
            total INTEGER NOT NULL,
            updated_at TEXT NOT NULL
        )
    """)

    baris_baru = sumber.execute(
        "SELECT id, pelanggan, total, updated_at FROM pesanan "
        "WHERE updated_at > ? ORDER BY updated_at",
        (watermark,),
    ).fetchall()

    for baris in baris_baru:
        tujuan.execute("INSERT OR REPLACE INTO pesanan VALUES (?, ?, ?, ?)", baris)
    tujuan.commit()

    if baris_baru:
        simpan_watermark(baris_baru[-1][3])

    total_gudang = tujuan.execute("SELECT COUNT(*) FROM pesanan").fetchone()[0]
    print(f"Baris baru diambil  : {len(baris_baru)}")
    print(f"Watermark baru      : {baca_watermark()}")
    print(f"Total baris di gudang: {total_gudang}")
    sumber.close()
    tujuan.close()


jalankan()

Dijalankan dua kali berturut-turut, hasilnya:

Output yang Dihasilkan:

Watermark awal      : 1970-01-01T00:00:00
Baris baru diambil  : 6
Watermark baru      : 2026-10-03T14:10:00
Total baris di gudang: 6

Run kedua sesaat kemudian:

Output yang Dihasilkan:

Watermark awal      : 2026-10-03T14:10:00
Baris baru diambil  : 0
Watermark baru      : 2026-10-03T14:10:00
Total baris di gudang: 6

Mari kita telaah baris-baris kode di atas:

1. Baris 5-6 (BERKAS_WATERMARK dan WATERMARK_AWAL): file state dideklarasikan sebagai konstanta, dengan nilai awal tahun 1970 agar run pertama menganggap seluruh tabel belum pernah dibaca.
2. Baris 9-12 (baca_watermark): fungsi mengembalikan watermark tersimpan bila file ada; jika tidak, nilai awal yang dikembalikan — skrip jadi aman dijalankan untuk pertama kali.
3. Baris 36-40 (SELECT ... WHERE updated_at > ?): query hanya memilih baris setelah watermark. Parameter ? mencegah injeksi SQL dan tetap benar karena watermark berformat ISO yang terurut secara leksikografis.
4. Baris 42-44 (INSERT OR REPLACE): baris dimasukkan dengan OR REPLACE sehingga pengiriman ulang memperbarui baris yang sama alih-alih menduplikasinya — sifat idempotent.
5. Baris 46-47 (simpan_watermark(...)): watermark hanya diperbarui setelah tujuan.commit() berhasil, dan diambil dari baris terakhir hasil ORDER BY updated_at — posisi paling depan yang sudah diproses.

Membuktikan siklusnya, kita tambahkan dua pesanan baru bertanggal 4 Oktober ke sumber.db, lalu kita jalankan skrip yang sama untuk ketiga kalinya:

Output yang Dihasilkan:

Watermark awal      : 2026-10-03T14:10:00
Baris baru diambil  : 2
Watermark baru      : 2026-10-04T10:30:00
Total baris di gudang: 8

Tiga run tersebut memperlihatkan siklus lengkap incremental: run pertama memuat 6 baris, run kedua menemukan 0 baris baru, dan run ketiga hanya mengambil 2 baris yang baru ditambahkan. Rincian modul json dan sqlite3 tersedia di dokumentasi json dan dokumentasi sqlite3 .

Incremental bukan satu-satunya strategi — tabel kecil seperti daftar produk justru lebih sederhana disalin penuh setiap hari. Kapan memilih yang mana kita bedah tuntas di chapter Full Load vs Incremental Load.