Toko online kita kini mencatat puluhan ribu pesanan setiap hari, dan tabel pesanan terus membesar. Chapter Join dan Agregasi Data sudah menunjukkan cara membangun agregat penjualan harian, tetapi menyisakan satu pertanyaan operasional: apakah tabel agregat harus dihitung ulang dari seluruh riwayat pesanan setiap kali pipeline berjalan? Incremental transformation menjawab pertanyaan tersebut dengan hanya memproses data baru sejak eksekusi terakhir, lalu menggabungkan hasilnya ke tabel agregat yang sudah ada sehingga biaya compute tetap konstan seiring pertumbuhan data.

Masalah Transform Ulang Seluruh Data #

Full recompute mudah ditulis: setiap malam, baca semua baris pesanan sejak hari pertama toko berdiri, lalu tulis ulang tabel agregat. Masalahnya, biaya pendekatan tersebut tumbuh seiring waktu. Dengan riwayat dua tahun sekitar 40 juta baris dan pertumbuhan 150 ribu baris per hari, pipeline memproses ulang 40 juta baris padahal hanya 0,4% di antaranya baru. Sumber daya compute terpakai untuk 99,6% data lama yang hasil agregasinya tidak berubah.

Incremental transformation membalik logikanya: proses hanya delta (baris baru sejak eksekusi terakhir), agregasikan delta tersebut, lalu merge hasilnya ke tabel agregat. Volume yang diproses setiap eksekusi sepadan dengan laju data baru, bukan dengan panjang riwayat.

Watermark sebagai Penanda Batas Pemrosesan #

Delta dibatasi oleh watermark: nilai terbesar dari kolom penanda (biasanya timestamp) yang sudah diproses. Watermark disimpan di tabel state terpisah agar tetap bertahan antar eksekusi job. Query pengambil delta cukup menyaring WHERE waktu > watermark, sehingga baris lama tidak pernah dibaca ulang.

Pemilihan kolom watermark menentukan karakter pipeline. Kolom berbasis waktu penyimpanan (kapan baris masuk ke sumber) menjaga delta tetap kecil, tetapi mengabaikan keterlambatan data. Kolom berbasis waktu kejadian lebih jujur terhadap makna bisnis, namun menuntut penanganan khusus bila baris lama tiba belakangan — pembahasan lengkapnya ada pada Chapter Late Arriving Data. Dua properti yang wajib dipenuhi kolom mana pun: nilai yang monoton naik dan terindeks, agar query delta tetap murah di tabel besar.

Implementasi: Transform Delta dan Merge ke Tabel Agregat #

Kita memakai tiga tabel: pesanan sebagai sumber, penjualan_harian sebagai tabel agregat, dan etl_state sebagai penyimpan watermark. Fungsi transform_incremental membaca delta, mengagregasi per tanggal, lalu menulis hasil lewat klausa ON CONFLICT ... DO UPDATE milik SQLite — baris tanggal baru disisipkan, baris tanggal lama ditambah.

python 
import sqlite3
from pathlib import Path

DB = Path("toko.db")
if DB.exists():
    DB.unlink()

koneksi = sqlite3.connect(DB)
koneksi.executescript("""
CREATE TABLE pesanan (
    id INTEGER PRIMARY KEY,
    waktu TEXT NOT NULL,
    pelanggan TEXT NOT NULL,
    produk TEXT NOT NULL,
    qty INTEGER NOT NULL,
    harga INTEGER NOT NULL
);
INSERT INTO pesanan (waktu, pelanggan, produk, qty, harga) VALUES
    ('2026-01-10T09:15:00', 'Budi', 'Kemeja Batik', 1, 250000),
    ('2026-01-10T14:30:00', 'Siti', 'Kain Songket', 2, 180000),
    ('2026-01-10T19:45:00', 'Dewi', 'Blouse Katik', 1, 120000),
    ('2026-01-11T08:05:00', 'Budi', 'Seluar Linen', 1, 210000),
    ('2026-01-11T13:20:00', 'Rina', 'Kemeja Batik', 3, 250000),
    ('2026-01-11T21:10:00', 'Siti', 'Rok Tenun', 2, 150000);
CREATE TABLE penjualan_harian (
    tanggal TEXT PRIMARY KEY,
    jumlah_pesanan INTEGER NOT NULL,
    total_pendapatan INTEGER NOT NULL
);
CREATE TABLE etl_state (
    nama_job TEXT PRIMARY KEY,
    watermark TEXT NOT NULL
);
INSERT INTO etl_state VALUES ('agg_harian', '0000-01-01T00:00:00');
""")


def transform_incremental(koneksi):
    (watermark_lama,) = koneksi.execute(
        "SELECT watermark FROM etl_state WHERE nama_job = 'agg_harian'"
    ).fetchone()
    baris_baru = koneksi.execute(
        "SELECT waktu, qty, harga FROM pesanan WHERE waktu > ? ORDER BY waktu",
        (watermark_lama,),
    ).fetchall()
    if not baris_baru:
        print(f"  tidak ada data baru, watermark tetap {watermark_lama}")
        return

    agregat = {}
    for waktu, qty, harga in baris_baru:
        tanggal = waktu[:10]
        jumlah, total = agregat.get(tanggal, (0, 0))
        agregat[tanggal] = (jumlah + 1, total + qty * harga)

    for tanggal, (jumlah, total) in sorted(agregat.items()):
        koneksi.execute(
            """INSERT INTO penjualan_harian (tanggal, jumlah_pesanan, total_pendapatan)
               VALUES (?, ?, ?)
               ON CONFLICT(tanggal) DO UPDATE SET
                   jumlah_pesanan = jumlah_pesanan + excluded.jumlah_pesanan,
                   total_pendapatan = total_pendapatan + excluded.total_pendapatan""",
            (tanggal, jumlah, total),
        )
        print(f"  merge {tanggal}: +{jumlah} pesanan, +Rp{total}")

    watermark_baru = baris_baru[-1][0]
    koneksi.execute(
        "UPDATE etl_state SET watermark = ? WHERE nama_job = 'agg_harian'",
        (watermark_baru,),
    )
    koneksi.commit()
    print(f"  watermark maju: {watermark_lama} -> {watermark_baru}")


def cek_konsistensi(koneksi):
    inkremental = dict((t, (j, tot)) for t, j, tot in koneksi.execute(
        "SELECT tanggal, jumlah_pesanan, total_pendapatan FROM penjualan_harian"))
    penuh = dict((t, (j, tot)) for t, j, tot in koneksi.execute(
        """SELECT substr(waktu, 1, 10), COUNT(*), SUM(qty * harga)
           FROM pesanan GROUP BY substr(waktu, 1, 10)"""))
    hasil = "KONSISTEN" if inkremental == penuh else "TIDAK KONSISTEN"
    print(f"  hasil incremental vs full recompute: {hasil}")


print("=== RUN 1: muat awal ===")
transform_incremental(koneksi)

koneksi.executemany(
    "INSERT INTO pesanan (waktu, pelanggan, produk, qty, harga) VALUES (?, ?, ?, ?, ?)",
    [
        ("2026-01-11T22:59:59", "Dewi", "Kain Songket", 1, 180000),
        ("2026-01-12T07:40:00", "Rina", "Blouse Katun", 2, 120000),
        ("2026-01-12T11:25:00", "Budi", "Rok Tenun", 1, 150000),
    ],
)
koneksi.commit()

print("=== RUN 2: data tambahan (termasuk hari lama) ===")
transform_incremental(koneksi)

print("=== RUN 3: tanpa data baru ===")
transform_incremental(koneksi)

print("=== VERIFIKASI ===")
cek_konsistensi(koneksi)

print("=== TABEL penjualan_harian ===")
for tanggal, jumlah, total in koneksi.execute(
    "SELECT tanggal, jumlah_pesanan, total_pendapatan FROM penjualan_harian ORDER BY tanggal"
):
    print(f"  {tanggal} | {jumlah} pesanan | Rp{total}")

koneksi.close()

Output yang Dihasilkan:

text 
=== RUN 1: muat awal ===
  merge 2026-01-10: +3 pesanan, +Rp730000
  merge 2026-01-11: +3 pesanan, +Rp1260000
  watermark maju: 0000-01-01T00:00:00 -> 2026-01-11T21:10:00
=== RUN 2: data tambahan (termasuk hari lama) ===
  merge 2026-01-11: +1 pesanan, +Rp180000
  merge 2026-01-12: +2 pesanan, +Rp390000
  watermark maju: 2026-01-11T21:10:00 -> 2026-01-12T11:25:00
=== RUN 3: tanpa data baru ===
  tidak ada data baru, watermark tetap 2026-01-12T11:25:00
=== VERIFIKASI ===
  hasil incremental vs full recompute: KONSISTEN
=== TABEL penjualan_harian ===
  2026-01-10 | 3 pesanan | Rp730000
  2026-01-11 | 4 pesanan | Rp1440000
  2026-01-12 | 2 pesanan | Rp390000

Mari kita telaah baris-baris kode di atas:

  1. Baris 39-48 (transform_incremental, bagian baca): fungsi mengambil watermark lama dari etl_state, lalu memilih hanya baris dengan waktu lebih besar dari watermark tersebut. Bila delta kosong, fungsi berhenti tanpa menyentuh tabel agregat — inilah perilaku RUN 3.
  2. Baris 50-54 (agregasi delta): setiap baris delta diakumulasi ke dictionary agregat per tanggal; potongan waktu[:10] mengambil bagian tanggal dari timestamp ISO.
  3. Baris 56-65 (merge ke tabel agregat): klausa ON CONFLICT(tanggal) DO UPDATE menjalankan dua cabang — tanggal yang belum ada di penjualan_harian disisipkan sebagai baris baru, sedangkan tanggal yang sudah ada nilainya ditambah dengan nilai delta dari excluded.
  4. Baris 67-73 (kemajuan watermark): watermark baru diambil dari waktu baris delta terakhir, lalu koneksi.commit() menutup transaksi sehingga penulisan agregat dan kemajuan watermark tersimpan bersamaan.

Perhatikan RUN 2: satu pesanan baru jatuh pada tanggal 2026-01-11 yang sudah ada di tabel agregat, sehingga baris tersebut bertambah dari 3 menjadi 4 pesanan dan totalnya naik dari Rp1260000 menjadi Rp1440000 — bukan menimpa, melainkan bergabung. Tanggal 2026-01-12 sebaliknya langsung terbentuk sebagai partisi baru. Bagian VERIFIKASI kemudian membandingkan isi tabel incremental dengan agregat yang dihitung penuh dari sumber, dan keduanya cocok.

Detail perilaku ON CONFLICT ... DO UPDATE terdokumentasi resmi di SQLite UPSERT . Pola merge-tambah di atas benar hanya jika setiap delta diproses tepat satu kali; begitu job gagal di tengah jalan lalu dijalankan ulang, delta yang sama berisiko dihitung dua kali. Chapter Idempotent Transformation memperkenalkan pola tulis yang aman diulang untuk menutup celah tersebut.