Incremental Models

Mart fct_pesanan dari Chapter Model Staging, Intermediate, dan Marts ditulis ulang penuh setiap dbt run: dbt membangun tabel baru dari seluruh riwayat, lalu menggantikan tabel lama. Untuk ribuan baris, langkah tersebut selesai dalam hitungan detik; untuk ratusan juta baris pesanan, menulis ulang semuanya setiap hari berarti membayar compute untuk data yang tidak berubah. dbt menyediakan materialization khusus untuk kondisi ini: incremental model, yang hanya memproses delta sejak run terakhir. Pola umumnya — watermark, delta, merge — sudah kita bedah pada Chapter Incremental Transformation; chapter ini membedah versi dbt-nya beserta jebakan yang menyertainya.

Konfigurasi Model Incremental #

Model incremental adalah model biasa dengan tiga konfigurasi tambahan:

{{ config(
    materialized='incremental',
    unique_key='id_pesanan',
    incremental_strategy='merge'
) }}

select
    id_pesanan, email, waktu_kejadian, diperbarui_pada,
    qty * harga as total
from {{ source('aplikasi_toko', 'pesanan') }}

{% if is_incremental() %}
where diperbarui_pada > (select max(diperbarui_pada) from {{ this }})
{% endif %}

Mari kita telaah baris-baris model di atas:

1. Baris 1-5 (config): materialized='incremental' memberi tahu dbt untuk tidak menulis ulang tabel penuh; unique_key menjadi kolom pencocokan antara data baru dan tabel lama; incremental_strategy='merge' menentukan cara dbt menggabungkan keduanya.
2. Baris 10 (from source(...)): membaca sumber raw.pesanan yang dideklarasikan pada Chapter dbt.
3. Baris 12-14 (is_incremental()): blok kondisional Jinja. Saat tabel target belum ada — run pertama — macro bernilai false dan filter tidak ikut dikompilasi, sehingga dbt menjalankan SELECT penuh dan membuat tabel. Run berikutnya macro bernilai true; {{ this }} merujuk tabel model itu sendiri, sehingga filter hanya mengambil baris dengan diperbarui_pada lebih baru daripada yang sudah dimiliki target.

Perhatikan bahwa filter memakai kolom diperbarui_pada — waktu kedatangan pembaruan di sumber — bukan waktu_kejadian transaksinya. Perbedaan dua kolom ini menjadi penting sebentar lagi.

Simulasi Empat Run di DuckDB #

dbt tidak terpasang di lingkungan bab ini, jadi kita tiru perilakunya di DuckDB: skrip berikut menjalankan SQL yang setara dengan hasil kompilasi dbt untuk model di atas — CREATE TABLE penuh untuk run pertama, lalu MERGE delta untuk run berikutnya. Blok di bawah memotong bagian awal skrip yang membuat tabel sumber sumber_pesanan berisi tiga pesanan:

# --- RUN 1: tabel target belum ada -> dbt menjalankan SELECT penuh ---
con.execute("""
CREATE TABLE fct_pesanan AS
SELECT id_pesanan, email, waktu_kejadian, diperbarui_pada, qty * harga AS total
FROM sumber_pesanan
""")
print("=== RUN 1 (full build, jalur is_incremental() == false) ===")
print("  baris target:", con.execute("SELECT COUNT(*) FROM fct_pesanan").fetchone()[0])

# --- RUN 2: koreksi ORD-2 (qty 2 -> 3) + ORD-4 baru; filter pakai diperbarui_pada ---
con.execute("""
UPDATE sumber_pesanan
SET qty = 3, diperbarui_pada = TIMESTAMP '2026-01-13 08:35'
WHERE id_pesanan = 'ORD-2';
INSERT INTO sumber_pesanan VALUES
    ('ORD-4', '[email protected]', TIMESTAMP '2026-01-13 08:40', TIMESTAMP '2026-01-13 08:40', 2, 150000);
""")

print()
print("=== RUN 2 (MERGE delta, filter diperbarui_pada > max) ===")
con.execute("""
MERGE INTO fct_pesanan AS t
USING (
    SELECT id_pesanan, email, waktu_kejadian, diperbarui_pada, qty * harga AS total
    FROM sumber_pesanan
    WHERE diperbarui_pada > (SELECT max(diperbarui_pada) FROM fct_pesanan)
) AS d ON t.id_pesanan = d.id_pesanan
WHEN MATCHED THEN UPDATE SET total = d.total, diperbarui_pada = d.diperbarui_pada
WHEN NOT MATCHED THEN
    INSERT (id_pesanan, email, waktu_kejadian, diperbarui_pada, total)
    VALUES (d.id_pesanan, d.email, d.waktu_kejadian, d.diperbarui_pada, d.total)
""")
print("  ORD-2 total :", con.execute(
    "SELECT total FROM fct_pesanan WHERE id_pesanan = 'ORD-2'").fetchone()[0],
    "(koreksi 2 -> 3 item)")
print("  ORD-4 total :", con.execute(
    "SELECT total FROM fct_pesanan WHERE id_pesanan = 'ORD-4'").fetchone()[0],
    "(baris baru)")


def cek_konsistensi():
    a = dict(con.execute("SELECT id_pesanan, total FROM fct_pesanan").fetchall())
    b = dict(con.execute(
        "SELECT id_pesanan, qty * harga FROM sumber_pesanan"
    ).fetchall())
    return ("KONSISTEN" if a == b else "TIDAK KONSISTEN"), a


print("  verifikasi vs full recompute:", cek_konsistensi()[0])

Output yang Dihasilkan:

=== RUN 1 (full build, jalur is_incremental() == false) ===
  baris target: 3

=== RUN 2 (MERGE delta, filter diperbarui_pada > max) ===
  ORD-2 total : 540000 (koreksi 2 -> 3 item)
  ORD-4 total : 300000 (baris baru)
  verifikasi vs full recompute: KONSISTEN

Mari kita telaah bagian-bagian penting skrip di atas:

1. Blok RUN 1: jalur is_incremental() == false — seluruh sumber dibaca, tabel target dibuat dengan tiga baris.
2. Blok RUN 2 (UPDATE + INSERT): mensimulasikan sumber yang berubah sejak run pertama: koreksi qty ORD-2 dari 2 menjadi 3, plus satu pesanan baru ORD-4.
3. Blok MERGE INTO: subquery delta menyaring diperbarui_pada > max(diperbarui_pada) milik target, sehingga hanya ORD-2 dan ORD-4 yang diproses. WHEN MATCHED memperbarui ORD-2 menjadi 540000; WHEN NOT MATCHED menyisipkan ORD-4.
4. Fungsi cek_konsistensi: membandingkan isi target dengan perhitungan ulang penuh dari sumber — cara paling jujur memastikan jalur incremental tidak menyimpang dari hasil full rebuild.

Jebakan Late-Arriving Data #

RUN 3 menguji skenario yang berbeda: pesanan ORD-0 terjadi 11 Januari pukul 22:59, tetapi baru tiba di sumber pada 13 Januari pukul 09:10 — ponsel kasir yang baru tersinkron, persis kasus Chapter Late Arriving Data. Skrip yang sama menambahkan ORD-0 ke sumber, lalu menjalankan MERGE dengan dua filter berbeda: RUN 3 memakai waktu_kejadian secara naif, RUN 4 memakai jendela lookback:

-- RUN 3: filter naif pada waktu kejadian
where waktu_kejadian > (select max(waktu_kejadian) from fct_pesanan)

-- RUN 4: jendela lookback tiga hari
where waktu_kejadian > (select max(waktu_kejadian) from fct_pesanan) - interval 3 day

Output yang Dihasilkan:

=== RUN 3 (late data ORD-0, filter naif waktu_kejadian > max) ===
  ORD-0 ada di target? : False
  verifikasi vs full recompute: TIDAK KONSISTEN

=== RUN 4 (filter lookback: waktu_kejadian > max - INTERVAL 3 DAY) ===
  ORD-0 ada di target? : True
  verifikasi vs full recompute: KONSISTEN
  ('ORD-0', '2026-01-11 22:59', 99000)
  ('ORD-1', '2026-01-10 09:15', 250000)
  ('ORD-2', '2026-01-11 14:30', 540000)
  ('ORD-3', '2026-01-12 11:25', 120000)
  ('ORD-4', '2026-01-13 08:40', 300000)

RUN 3 gagal senyap: max(waktu_kejadian) di target sudah menunjuk ORD-4 (13 Januari 08:40), sedangkan ORD-0 terjadi 11 Januari — filter menolaknya tanpa pesan galat, dan verifikasi yang menangkap drift tersebut. RUN 4 membaca ulang tiga hari terakhir; karena MERGE mencocokkan unique_key, baris lama yang ikut terbaca hanya ditimpa dengan nilai yang sama, sementara ORD-0 akhirnya masuk. Jendela lookback adalah kompromi: terlalu sempit masih membocorkan data telat, terlalu lebar memperbesar delta yang diproses setiap run.

Strategi Incremental dan unique_key #

merge hanyalah satu dari beberapa strategi bawaan dbt; dokumentasi incremental models mencatat strategi mana yang didukung adapter kita:

Peran unique_key berbeda antar strategi: merge dan delete+insert memakainya sebagai kunci pencocokan, sedangkan append tidak membutuhkannya sama sekali — rinciannya ada di referensi unique_key . Untuk kondisi darurat, flag dbt run --full-refresh memaksa jalur penuh: tabel dibangun ulang dari nol, berguna setelah logika model berubah atau schema target berubah.

Incremental model menjaga target selalu mutakhir: koreksi menimpa nilai lama, dan nilai lama hilang begitu saja. Namun ada pertanyaan yang justru membutuhkan nilai lama tersebut — di kota mana Budi tinggal saat pesanan Januari dibuat, sebelum pindah bulan berikutnya? Kebutuhan riwayat perubahan seperti ini ditangani fitur dbt yang berbeda, materi Chapter Snapshots.