Belajar Data Engineer - Data Ingestion & CDC

#Pendahuluan

Semua pipeline dimulai dari satu pertanyaan: bagaimana data masuk ke platform? Di episode 5 kalian menarik data dari API dengan Python, dan di episode 11 dari event stream. Episode ini membawa ingestion ke level profesional: kapan memakai full load, kapan incremental, dan bagaimana Change Data Capture (CDC) memungkinkan data dari database produksi hadir di warehouse hampir seketika.

Mengapa topik ini penting? Karena metode ingestion menentukan latency, beban pada sumber, dan biaya. Memuat ulang 2 juta baris setiap 5 menit hanya untuk 100 baris yang berubah adalah pemborosan — dan di sisi lain, melewatkan perubahan yang tidak terdeteksi adalah kehilangan data. Data engineer yang baik memilih metode yang tepat per sumber.

#Full Load vs Incremental Load

#Full Load

Memuat seluruh data dari sumber setiap kali. Karakteristiknya:

• Sederhana: tanpa tracking posisi, tanpa state.
• Murah di awal: cocok untuk tabel kecil (puluhan ribu baris).
• Tidak scalable: tabel milyaran baris tidak bisa di-full-load tiap malam.

#Incremental Load

Hanya memuat data yang berubah sejak ekstraksi terakhir. Membutuhkan penanda (watermark) untuk melacak posisi. Tiga pendekatan umum:

-- Simpan state watermark di tabel kecil
CREATE TABLE IF NOT EXISTS etl_state (
    source TEXT PRIMARY KEY,
    last_run TIMESTAMP
);
 
INSERT INTO staging.orders_inc
SELECT * FROM public.orders
WHERE updated_at > COALESCE(
    (SELECT last_run FROM etl_state WHERE source = 'orders'),
    '1970-01-01'
);
 
UPDATE etl_state SET last_run = now() WHERE source = 'orders';

#Change Data Capture (CDC): Membaca Transaction Log

Pendekatan timestamp punya dua kelemahan serius: data yang dihapus tidak terdeteksi, dan kolom updated_at sering tidak akurat. CDC menyelesaikan keduanya dengan cara yang berbeda — ia membaca transaction log database (WAL di PostgreSQL, binlog di MySQL) yang mencatat setiap perubahan: insert, update, dan delete.

Keuntungan CDC:

• Lengkap: menangkap delete dan perubahan apa pun, tanpa bergantung kolom updated_at.
• Latency rendah & non-intrusif: perubahan diteruskan dalam detik tanpa menambah beban query pada database produksi.

#Debezium: Connector CDC Standar

Debezium adalah platform CDC open-source yang membaca transaction log dan menerbitkan setiap perubahan sebagai event ke Kafka. Topologi klasiknya:

Menjalankan Debezium untuk PostgreSQL dengan Kafka Connect:

version: "3"
services:
  connect:
    image: quay.io/debezium/connect:3.0
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_status
    ports:
      - "8083:8083"

curl -i -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "pg-orders-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "debezium",
      "database.password": "dbz",
      "database.dbname": "de_lab",
      "topic.prefix": "debezium",
      "plugin.name": "pgoutput",
      "slot.name": "debezium_slot"
    }
  }'

Setiap perubahan pada tabel orders kini muncul sebagai event JSON di topic debezium.public.orders — lengkap dengan metadata operasi:

{
  "op": "u",
  "before": { "order_id": 42, "status": "pending", "amount": 25000 },
  "after": { "order_id": 42, "status": "completed", "amount": 25000 }
}

Field op menandai operasi: c (create), u (update), d (delete). before/after memuat nilai lama dan baru. Inilah bahan mentah untuk sinkronisasi incremental yang akurat.

#Menerima Event CDC di Warehouse

Consumer membaca event CDC dan menerapkannya ke tabel warehouse — biasanya dengan MERGE di lakehouse (episode 10):

from confluent_kafka import Consumer
import json, psycopg2
 
c = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "cdc-warehouse",
    "auto.offset.reset": "earliest",
})
c.subscribe(["debezium.public.orders"])
 
conn = psycopg2.connect(host="localhost", user="de", password="secret", dbname="de_lab")
 
while True:
    msg = c.poll(1.0)
    if msg is None:
        continue
    evt = json.loads(msg.value())
    op, after = evt.get("op"), evt.get("after")
    if op == "c":
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO silver.orders VALUES (%s,%s,%s,%s)",
                (after["order_id"], after["status"], after["amount"], after["created_at"]),
            )
            conn.commit()
    elif op == "d":
        with conn.cursor() as cur:
            cur.execute("DELETE FROM silver.orders WHERE order_id = %s", (evt["before"]["order_id"],))
            conn.commit()

#Kapan Memakai Apa: Keputusan Ingestion

Kerangka keputusan yang bisa kalian pakai:

1. Tabel kecil (kecil = bisa full load murah) → full load terjadwal, paling sederhana.
2. Tabel besar dengan updated_at yang andal → incremental berbasis timestamp.
3. Tabel besar, butuh delete tracking, atau latency rendah → CDC dengan Debezium.
4. Data dari SaaS tanpa akses DB → API + incremental (episode 5), atau connector Airbyte/Fivetran (episode 6).

#Kesalahan Umum (Common Pitfalls)

1. Full load untuk tabel besar. Biaya dan beban sumber tidak sebanding dengan manfaatnya. Pindah ke incremental.

2. Timestamp watermark tanpa update kolom. Jika aplikasi tidak mengupdate updated_at, baris berubah tapi tak terdeteksi. Verifikasi integritas watermark.

3. CDC tanpa penanganan delete. Hanya mengerjakan insert/update tapi mengabaikan op: d membuat data warehouse menahan data yang sudah dihapus di sumber.

4. Replication slot terlupakan. WAL membengkak, disk penuh, database produksi berhenti. Monitor slot sejak hari pertama.

#Penutup

Di episode 12 ini kalian telah menguasai ingestion profesional:

• Full load sederhana tapi tidak scalable; incremental wajib untuk data besar.
• Pendekatan incremental: timestamp-based, diff, dan CDC — yang terakhir paling lengkap & real-time.
• Debezium membaca transaction log dan menerbitkan perubahan ke Kafka sebagai event dengan op/before/after.
• Consumer menerapkan event CDC ke warehouse dengan insert/update/delete — sinkron dalam detik.

Di episode 13 selanjutnya kita naik level orkestrasi: orkestrasi & scheduling lanjutan — Airflow advanced (sensors, retries, backfills), perbandingan dengan Dagster dan Prefect, dan praktik membangun pipeline production-grade. Sampai jumpa di episode 13!