Pipeline batch harian kita membaca ulang seluruh partisi setiap kali berjalan. Dengan riwayat tiga hari, kebiasaan tersebut tidak terasa; dengan riwayat dua tahun dan 40 juta baris, pipeline akan menghabiskan 99 persen waktunya untuk memproses data lama yang hasilnya tidak pernah berubah. Pipeline incremental membalik pendekatan tersebut: setiap eksekusi hanya memproses data baru atau data yang berubah sejak eksekusi terakhir, lalu menggabungkan hasilnya ke tabel target. Chapter ini merakit pipeline incremental tingkat baris, dengan watermark yang disimpan pada state file dan merge berbasis upsert.

Watermark dan State Pipeline #

Watermark adalah batas atas nilai kolom penanda yang sudah selesai diproses. Pipeline menyimpan watermark pada berkas state, misalnya state.json, sehingga nilai tersebut bertahan antareksekusi. Query pengambil delta tinggal menyaring baris dengan nilai penanda di atas watermark, dan baris lama tidak pernah dibaca ulang.

Pemilihan kolom penanda menentukan kemampuan pipeline. Penanda berupa order_id yang selalu naik cukup untuk menangkap pesanan baru, tetapi tidak untuk menangkap perubahan status pesanan lama, karena nilai order_id pesanan lama tidak pernah berubah. Penanda berupa updated_at menyelesaikan kedua kasus tersebut: setiap perubahan status menaikkan nilai updated_at, sehingga baris yang berubah ikut terbaca sebagai delta. Dua syarat wajibnya sama: nilai penanda harus monoton naik dan terindeks supaya query delta tetap murah.

Merge ke Target dengan Upsert #

Baris delta tidak boleh ditambah begitu saja ke tabel target, karena baris dengan order_id sama bisa muncul kembali sebagai perubahan. Merge mengatasi hal tersebut lewat klausa INSERT ... ON CONFLICT(order_id) DO UPDATE: bila order_id belum ada di target, baris disisipkan; bila sudah ada, kolom yang berubah diperbarui. Tabel target dengan demikian selalu mencerminkan kondisi terbaru sumber tanpa duplikat.

Contoh: Tiga Run dalam Sehari #

Skenario kita memakai dua database: operational.db sebagai sumber (database transaksi toko) dan analytics.db sebagai target, dengan state.json sebagai penyimpan watermark. Pipeline dijalankan tiga kali. Antara run, kita menyisipkan pesanan baru dan mengubah status satu pesanan lama agar ketiga situasi terlihat: muat awal, pesanan baru, dan perubahan status.

python 
"""Pipeline incremental TokoKita: watermark di state file, merge upsert ke target."""
import json
import sqlite3
from pathlib import Path

SUMBER = Path("operational.db")
TARGET = Path("analytics.db")
STATE = Path("state.json")


def siapkan_sumber():
    for berkas in (SUMBER, TARGET, STATE):
        if berkas.exists():
            berkas.unlink()
    k = sqlite3.connect(SUMBER)
    k.executescript("""
CREATE TABLE orders (
    order_id INTEGER PRIMARY KEY, pelanggan TEXT, email TEXT, tanggal TEXT,
    status TEXT, total INTEGER, updated_at TEXT NOT NULL);
INSERT INTO orders VALUES
 (1, 'Budi', '[email protected]', '2026-01-05', 'diproses', 450000, '2026-01-05T10:12:00'),
 (2, 'Siti', '[email protected]', '2026-01-05', 'dikirim', 275000, '2026-01-05T11:40:00'),
 (3, 'Dewi', '[email protected]', '2026-01-06', 'diproses', 300000, '2026-01-06T09:05:00'),
 (4, 'Andi', '[email protected]', '2026-01-06', 'selesai', 825000, '2026-01-06T13:30:00'),
 (5, 'Rina', '[email protected]', '2026-01-06', 'dikirim', 400000, '2026-01-06T16:20:00');
""")
    k.commit()
    k.close()
    t = sqlite3.connect(TARGET)
    t.execute("""CREATE TABLE orders (
    order_id INTEGER PRIMARY KEY, pelanggan TEXT, email TEXT, tanggal TEXT,
    status TEXT, total INTEGER, updated_at TEXT)""")
    t.commit()
    t.close()


def tambah_pesanan(rows):
    k = sqlite3.connect(SUMBER)
    k.executemany("INSERT INTO orders VALUES (?,?,?,?,?,?,?)", rows)
    k.commit()
    k.close()


def ubah_status(order_id, status_baru, updated_at):
    k = sqlite3.connect(SUMBER)
    k.execute("UPDATE orders SET status = ?, updated_at = ? WHERE order_id = ?",
              (status_baru, updated_at, order_id))
    k.commit()
    k.close()


def jalankan_pipeline():
    state = (json.loads(STATE.read_text()) if STATE.exists()
             else {"watermark": "0000-01-01T00:00:00"})
    k = sqlite3.connect(SUMBER)
    delta = k.execute(
        "SELECT * FROM orders WHERE updated_at > ? ORDER BY updated_at, order_id",
        (state["watermark"],)).fetchall()
    k.close()
    if not delta:
        print("  tidak ada delta, target tidak disentuh")
        return
    t = sqlite3.connect(TARGET)
    for baris in delta:
        t.execute("""
            INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(order_id) DO UPDATE SET
                status = excluded.status, updated_at = excluded.updated_at
        """, baris)
    t.commit()
    t.close()
    watermark_baru = delta[-1][6]
    STATE.write_text(json.dumps({"watermark": watermark_baru}, indent=2) + "\n")
    ids = ", ".join(str(b[0]) for b in delta)
    print(f"  delta diproses: pesanan {ids} ({len(delta)} baris)")
    print(f"  watermark: {state['watermark']} -> {watermark_baru}")


siapkan_sumber()
print("=== RUN 1: muat awal ===")
jalankan_pipeline()
tambah_pesanan([
    (6, "Budi", "[email protected]", "2026-01-07", "diproses", 650000, "2026-01-07T10:15:00"),
    (7, "Siti", "[email protected]", "2026-01-07", "selesai", 150000, "2026-01-07T11:25:00"),
    (8, "Dewi", "[email protected]", "2026-01-07", "dikirim", 980000, "2026-01-07T14:50:00"),
    (9, "Andi", "[email protected]", "2026-01-07", "diproses", 300000, "2026-01-07T17:10:00"),
])
print("=== RUN 2: empat pesanan baru ===")
jalankan_pipeline()
ubah_status(3, "dikirim", "2026-01-08T08:45:00")
tambah_pesanan([
    (10, "Rina", "[email protected]", "2026-01-08", "diproses", 710000, "2026-01-08T09:30:00"),
    (11, "Budi", "[email protected]", "2026-01-08", "selesai", 260000, "2026-01-08T10:40:00"),
    (12, "Siti", "[email protected]", "2026-01-08", "dikirim", 540000, "2026-01-08T12:05:00"),
])
print("=== RUN 3: tiga pesanan baru + satu perubahan status ===")
jalankan_pipeline()
print("=== VERIFIKASI: isi tabel target ===")
t = sqlite3.connect(TARGET)
for order_id, pelanggan, status, updated_at in t.execute(
        "SELECT order_id, pelanggan, status, updated_at FROM orders ORDER BY order_id"):
    print(f"  pesanan {order_id} | {pelanggan} | {status} | {updated_at}")
print(f"  total baris target: {t.execute('SELECT COUNT(*) FROM orders').fetchone()[0]}")
t.close()
print("=== isi state.json setelah RUN 3 ===")
print(STATE.read_text(), end="")

Output yang Dihasilkan:

text 
=== RUN 1: muat awal ===
  delta diproses: pesanan 1, 2, 3, 4, 5 (5 baris)
  watermark: 0000-01-01T00:00:00 -> 2026-01-06T16:20:00
=== RUN 2: empat pesanan baru ===
  delta diproses: pesanan 6, 7, 8, 9 (4 baris)
  watermark: 2026-01-06T16:20:00 -> 2026-01-07T17:10:00
=== RUN 3: tiga pesanan baru + satu perubahan status ===
  delta diproses: pesanan 3, 10, 11, 12 (4 baris)
  watermark: 2026-01-07T17:10:00 -> 2026-01-08T12:05:00
=== VERIFIKASI: isi tabel target ===
  pesanan 1 | Budi | diproses | 2026-01-05T10:12:00
  pesanan 2 | Siti | dikirim | 2026-01-05T11:40:00
  pesanan 3 | Dewi | dikirim | 2026-01-08T08:45:00
  pesanan 4 | Andi | selesai | 2026-01-06T13:30:00
  pesanan 5 | Rina | dikirim | 2026-01-06T16:20:00
  pesanan 6 | Budi | diproses | 2026-01-07T10:15:00
  pesanan 7 | Siti | selesai | 2026-01-07T11:25:00
  pesanan 8 | Dewi | dikirim | 2026-01-07T14:50:00
  pesanan 9 | Andi | diproses | 2026-01-07T17:10:00
  pesanan 10 | Rina | diproses | 2026-01-08T09:30:00
  pesanan 11 | Budi | selesai | 2026-01-08T10:40:00
  pesanan 12 | Siti | dikirim | 2026-01-08T12:05:00
  total baris target: 12
=== isi state.json setelah RUN 3 ===
{
  "watermark": "2026-01-08T12:05:00"
}

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

  1. Baris 53-54 (jalankan_pipeline, baca state): fungsi memuat state.json bila berkas ada; pada eksekusi pertama, berkas belum ada sehingga watermark dimulai dari nilai dasar 0000-01-01T00:00:00.
  2. Baris 56-58 (query delta): klausa WHERE updated_at > ? mengambil hanya baris dengan penanda di atas watermark, diurutkan berdasarkan updated_at lalu order_id agar urutan pemrosesan stabil.
  3. Baris 64-71 (merge upsert): tiap baris delta dijalankan melalui ON CONFLICT(order_id) DO UPDATE, sehingga pesanan yang sudah ada diperbarui, bukan disisipkan lagi.
  4. Baris 72-74 (kemajuan watermark): watermark baru diambil dari updated_at baris delta terakhir, lalu state.json ditulis sebelum fungsi melaporkan hasil.

Tiga baris keluaran di atas memuat bukti kerja incremental. RUN 2 hanya memproses pesanan 6 sampai 9; pesanan 1 sampai 5 tidak dibaca ulang karena berada di bawah watermark. RUN 3 memproses pesanan 3, 10, 11, 12: tiga pesanan baru ditambah satu perubahan status, karena ubah_status menaikkan updated_at pesanan 3 ke 2026-01-08T08:45:00. Verifikasi akhir menunjukkan pesanan 3 berstatus dikirim di target, dan total baris target tetap 12, bukan 13, karena merge menimpa baris lama alih-alih menambah baris baru.

AspekBatch PenuhIncremental
Volume per eksekusiSeluruh riwayat, tumbuh seiring waktuHanya delta, sepadan laju perubahan
Durasi eksekusiMembesar tiap bulanHampir konstan
Kebutuhan stateTidak adaWajib (watermark)
Perubahan data lamaTertangkap otomatisButuh penanda yang ikut berubah
Cocok untukBackfill, rekonsiliasi berkalaPipeline harian rutin

Perilaku ON CONFLICT ... DO UPDATE terdokumentasi resmi di SQLite UPSERT , dan rujukan lengkap SQL-nya ada di dokumentasi SQLite . Pipeline incremental kita kini berjalan sendiri, tetapi di dunia nyata pipeline jarang berdiri tunggal: pipeline analitik menunggu hasil pipeline ingestion, dan hubungan tersebut harus dikelola eksplisit. Kita bahas pada Chapter Dependency Management.