Change Data Capture

Database toko online berubah sepanjang waktu: harga produk dikoreksi, stok berkurang, dan pesanan lama dibatalkan. Pipeline yang kita bangun pada Chapter Incremental Ingestion biasanya mengandalkan kolom updated_at dan query berkala (polling) untuk menemukan baris baru. Pendekatan polling menanggung biaya query berat setiap menit, bisa melewatkan baris yang berubah dua kali di antara dua jeda polling, dan hampir mustahil mendeteksi baris yang dihapus. Change Data Capture (CDC) menyelesaikan ketiganya dengan cara membaca catatan perubahan langsung dari database, bukan bertanya berulang kali ke tabelnya.

Polling versus CDC #

Polling menanyakan isi tabel pada interval tetap, misalnya SELECT * FROM produk WHERE updated_at > :penanda_terakhir setiap lima menit. CDC bekerja dari arah sebaliknya: database sendiri yang mengumumkan setiap perubahan, dan pipeline hanya perlu membaca aliran pengumuman tersebut.

Log-Based CDC dan WAL PostgreSQL #

CDC paling andal tidak membaca tabel sama sekali, melainkan write-ahead log (WAL). PostgreSQL menulis setiap perubahan — INSERT, UPDATE, DELETE — ke WAL sebelum perubahan tersebut diterapkan ke tabel, seperti dijelaskan pada dokumentasi resmi di https://www.postgresql.org/docs/ . Fitur logical decoding PostgreSQL menerbitkan perubahan tersebut sebagai aliran event yang bisa dibaca pihak luar tanpa menambah beban pada tabel sumber.

Debezium adalah alat open source paling populer untuk memanfaatkan mekanisme tersebut. Debezium membaca WAL PostgreSQL (atau binlog MySQL, atau oplog MongoDB), mengubah setiap perubahan menjadi event terstruktur, lalu menyalurkan event ke Apache Kafka agar pipeline kita tinggal dikonsumsi sebagai topic. Dokumentasi lengkapnya tersedia di https://debezium.io/documentation/ . Dengan Debezium, pipeline menerima setiap perubahan dalam hitungan detik tanpa satu pun query tambahan ke tabel produksi.

Format Event CDC #

Setiap event CDC membawa tiga bagian kunci: op (jenis operasi), sebelum (nilai baris lama), dan sesudah (nilai baris baru). Debezium memakai kode operasi ringkas: c untuk create, u untuk update, d untuk delete, dan r untuk snapshot awal (read) yang menerbitkan seluruh isi tabel saat pertama kali terhubung. Berkat pasangan sebelum/sesudah, pipeline bisa menjawab pertanyaan seperti “harga berapa sebelum dikoreksi?” — sesuatu yang mustahil didapat dari snapshot tabel.

Simulasi CDC dengan Changelog SQLite #

Kita tidak butuh Kafka untuk memahami pola kerjanya. SQLite bisa meniru log-based CDC memakai trigger: setiap INSERT, UPDATE, dan DELETE pada tabel produk otomatis menulis satu baris event ke tabel produk_changelog, lengkap dengan kode op ala Debezium serta snapshot sebelum dan sesudah dalam format JSON. Tabel sumber tetap dijalankan seperti biasa oleh aplikasi toko.

import json
import sqlite3

sumber = sqlite3.connect(":memory:")
target = sqlite3.connect(":memory:")

# --- skema sumber: tabel bisnis + changelog + trigger ---
sumber.executescript("""
CREATE TABLE produk (
    id INTEGER PRIMARY KEY,
    nama TEXT NOT NULL,
    harga INTEGER NOT NULL
);
CREATE TABLE produk_changelog (
    seq INTEGER PRIMARY KEY AUTOINCREMENT,
    op TEXT NOT NULL,
    sebelum TEXT,
    sesudah TEXT,
    dipakai INTEGER DEFAULT 0
);
CREATE TRIGGER audit_insert AFTER INSERT ON produk BEGIN
    INSERT INTO produk_changelog(op, sebelum, sesudah)
    VALUES ('c', NULL,
            json_object('id', NEW.id, 'nama', NEW.nama, 'harga', NEW.harga));
END;
CREATE TRIGGER audit_update AFTER UPDATE ON produk BEGIN
    INSERT INTO produk_changelog(op, sebelum, sesudah)
    VALUES ('u',
            json_object('id', OLD.id, 'nama', OLD.nama, 'harga', OLD.harga),
            json_object('id', NEW.id, 'nama', NEW.nama, 'harga', NEW.harga));
END;
CREATE TRIGGER audit_delete AFTER DELETE ON produk BEGIN
    INSERT INTO produk_changelog(op, sebelum, sesudah)
    VALUES ('d',
            json_object('id', OLD.id, 'nama', OLD.nama, 'harga', OLD.harga),
            NULL);
END;
""")

# --- transaksi bisnis biasa di aplikasi toko ---
sumber.execute("INSERT INTO produk (nama, harga) VALUES ('Keyboard Mekanik', 750000)")
sumber.execute("INSERT INTO produk (nama, harga) VALUES ('Mouse Gaming', 350000)")
sumber.execute("UPDATE produk SET harga = 690000 WHERE nama = 'Keyboard Mekanik'")
sumber.execute("DELETE FROM produk WHERE nama = 'Mouse Gaming'")
sumber.commit()

print("event CDC yang tertangkap:")
for seq, op, sebelum, sesudah in sumber.execute(
    "SELECT seq, op, sebelum, sesudah FROM produk_changelog ORDER BY seq"
):
    print(f"  seq={seq} op={op} sebelum={sebelum} sesudah={sesudah}")

Output yang Dihasilkan:

event CDC yang tertangkap:
  seq=1 op=c sebelum=None sesudah={"id":1,"nama":"Keyboard Mekanik","harga":750000}
  seq=2 op=c sebelum=None sesudah={"id":2,"nama":"Mouse Gaming","harga":350000}
  seq=3 op=u sebelum={"id":1,"nama":"Keyboard Mekanik","harga":750000} sesudah={"id":1,"nama":"Keyboard Mekanik","harga":690000}
  seq=4 op=d sebelum={"id":2,"nama":"Mouse Gaming","harga":350000} sesudah=None

Mari kita telaah baris-baris kode di atas:

1. Trigger audit_insert: setiap INSERT baru menulis event op='c' dengan sesudah berisi nilai NEW — data baris yang baru dibuat — sementara sebelum diisi NULL karena baris memang belum ada.
2. Trigger audit_update: event op='u' menyimpan dua sisi sekaligus: OLD (harga 750000) sebagai sebelum dan NEW (harga 690000) sebagai sesudah.
3. Trigger audit_delete: event op='d' hanya menyimpan OLD pada kolom sebelum, karena setelah penghapusan tidak ada nilai baru.
4. Kolom dipakai INTEGER DEFAULT 0: penanda event mana yang sudah diterapkan pipeline — peran yang di sistem nyata dipegang oleh offset consumer Kafka.

Menerapkan Event ke Database Target #

Sisi consumer membaca event yang belum dipakai (dipakai = 0) secara berurutan, menerjemahkan setiap event menjadi operasi SQL di target, lalu menandai event tersebut sebagai selesai. Operasi c dan u diterapkan sebagai upsert agar event yang sampai dua kali tetap aman.

# --- terapkan event ke database target ---
target.execute("""
CREATE TABLE produk (
    id INTEGER PRIMARY KEY,
    nama TEXT NOT NULL,
    harga INTEGER NOT NULL
)
""")

event_baru = sumber.execute(
    "SELECT seq, op, sebelum, sesudah FROM produk_changelog "
    "WHERE dipakai = 0 ORDER BY seq"
).fetchall()

for seq, op, sebelum, sesudah in event_baru:
    if op in ("c", "u"):
        baris = json.loads(sesudah)
        target.execute(
            "INSERT INTO produk (id, nama, harga) VALUES (?, ?, ?) "
            "ON CONFLICT(id) DO UPDATE SET nama = excluded.nama, harga = excluded.harga",
            (baris["id"], baris["nama"], baris["harga"]),
        )
    elif op == "d":
        baris = json.loads(sebelum)
        target.execute("DELETE FROM produk WHERE id = ?", (baris["id"],))
    sumber.execute("UPDATE produk_changelog SET dipakai = 1 WHERE seq = ?", (seq,))

target.commit()
sumber.commit()
print("semua event diterapkan ke target")

print("isi tabel produk di target:")
for baris in target.execute("SELECT id, nama, harga FROM produk ORDER BY id"):
    print(" ", tuple(baris))

Output yang Dihasilkan:

semua event diterapkan ke target
isi tabel produk di target:
  (1, 'Keyboard Mekanik', 690000)

Hasil akhirnya tepat: target hanya memuat satu produk dengan harga terkoreksi 690000. Event u menimpa harga lama lewat klausa ON CONFLICT ... DO UPDATE, dan event d menghapus produk kedua sehingga target akhirnya identik dengan kondisi sumber. Query WHERE dipakai = 0 ORDER BY seq menjamin event diproses berurutan dan tidak ada yang terlewat — sifat yang sama ditawarkan offset pada Kafka.

CDC menuntaskan perubahan data dari database relasional. Namun sebagian besar sumber data toko online justru berupa HTTP API pihak ketiga — kurir, pembayaran, marketplace — yang menyajikan data dalam halaman-halaman. Cara menarik semua halaman tanpa ada data yang terlewat menjadi bahasan Chapter Menangani Pagination API.