DataOps nâng cao 23 — Ingestion & CDC nâng cao

Chữ E-L đứng trước chữ T

Trong ELT hiện đại, dbt lo chữ T (Transform) — biến dữ liệu thô đã nằm trong kho thành các bảng mô hình (dbt & ELT). Nhưng dbt không kéo dữ liệu từ nguồn về. Việc đó thuộc về E-L (Extract-Load): lấy dữ liệu ra khỏi hệ nguồn và nạp vào landing/raw zone. Đây là tầng ít được nói tới nhưng lại là nơi phần lớn sự cố dữ liệu bắt nguồn: nạp trùng làm nhân đôi số dư, bỏ sót bản ghi đến muộn, không bắt được lệnh xoá ở nguồn, hoặc schema nguồn đổi âm thầm làm gãy cả pipeline.

Bài này bàn về ingestion đáng tin dưới lăng kính DataOps: chọn đúng mẫu nạp, đảm bảo idempotent (chạy lại không sinh dữ liệu thừa), bắt được cả xoá/sửa qua CDC, và giữ raw zone bất biến để tái xử lý. Bối cảnh xuyên suốt là ngân hàng: đưa dữ liệu từ core banking (Oracle) vào lakehouse near-real-time mà vẫn đối soát được.

Ba mẫu nạp dữ liệu

Mọi cơ chế ingestion về cơ bản rơi vào ba mẫu, xếp theo độ phức tạp tăng dần.

Full load

Đơn giản nhất: SELECT * FROM nguồn, ghi đè toàn bộ bảng đích. Với bảng departments vài chục dòng thì hoàn hảo — chạy lại bao nhiêu lần cũng ra kết quả giống nhau (tự nhiên idempotent). Nhưng với bảng transactions hàng trăm triệu dòng, full load mỗi giờ là bất khả thi: nó nện I/O lên DB nguồn (đang phục vụ nghiệp vụ), tốn băng thông, và cửa sổ chạy dài dần theo dữ liệu.

Incremental theo watermark

Ý tưởng: lưu một watermark — giá trị lớn nhất của cột mốc đã nạp lần trước (thường là created_at, updated_at, hoặc một id tăng dần) — và lần sau chỉ kéo bản ghi có cột mốc lớn hơn watermark đó.

Hai cái bẫy kinh điển:

• Late-arriving data: bản ghi có created_at là thời điểm giao dịch phát sinh, nhưng chỉ tới kho muộn (do đồng bộ trễ, giao dịch offline được sync sau). Nếu watermark đã vượt qua thời điểm đó, bản ghi bị bỏ sót vĩnh viễn. Cách giảm thiểu: dùng cột updated_at thay vì created_at, và lùi cửa sổ một khoảng an toàn (ví dụ nạp lại 3 ngày gần nhất mỗi lần) để bắt bản ghi đến muộn.
• Cột mốc không đáng tin: nếu nguồn cho phép sửa updated_at bằng tay, hoặc backfill dữ liệu cũ với updated_at = now(), watermark sẽ nhảy loạn. Incremental giả định cột mốc đơn điệu tăng theo lần sửa — hệ nguồn phải bảo đảm điều đó, nếu không mẫu này sai âm thầm.

Điểm yếu chí mạng của incremental thuần: không bắt được DELETE. Xoá một dòng ở nguồn không đổi cột mốc nào cả, nên bản ghi đã xoá vẫn "sống" ở đích. Đây là lý do ngân hàng cần CDC.

CDC — Change Data Capture

CDC bắt mọi thay đổi (INSERT/UPDATE/DELETE) ở DB nguồn và biến chúng thành một luồng sự kiện (CDC & real-time, tổng quan Oracle CDC). Có ba biến thể:

• Log-based (khuyến nghị): đọc trực tiếp transaction log của DB — Oracle redo log/LogMiner, MySQL binlog, PostgreSQL WAL. Công cụ điển hình: Debezium (open-source), Oracle GoldenGate (thương mại). Ưu điểm quyết định: gần như không tải thêm lên nguồn (chỉ đọc log đã có sẵn), bắt được cả DELETE, và giữ đúng thứ tự thay đổi. Đây là lựa chọn chuẩn cho core banking.
• Trigger-based: gắn trigger vào bảng nguồn, mỗi thay đổi ghi vào một bảng "audit". Đơn giản nhưng thêm ghi vào mỗi transaction của nguồn — nặng, ít bên vận hành DB đồng ý.
• Query-based: chính là incremental theo updated_at ở trên. Gọi là "CDC" cho oai nhưng vẫn không bắt được DELETE.

Kiến trúc CDC → Kafka → lakehouse

Trong ngân hàng, luồng CDC log-based thường đi qua Kafka trước khi hạ cánh vào lakehouse:

Debezium đọc redo log của Oracle, sinh một sự kiện cho mỗi thay đổi, đẩy vào topic Kafka riêng theo bảng (kiến trúc Kafka). Kafka là bộ đệm bền, cho phép replay khi cần nạp lại. Consumer đọc từ Kafka, ghi thô vào raw zone (bất biến), rồi bước MERGE dedup và cập nhật bảng đích.

Một sự kiện Debezium (rút gọn, chỉ để minh hoạ — cấu trúc thật nhiều trường hơn):

{
  "op": "u",
  "ts_ms": 1720000000123,
  "before": { "id": 42, "balance": 1000000 },
  "after":  { "id": 42, "balance": 1500000 }
}

Trường op là chìa khoá: c (create), u (update), d (delete — cái mà incremental bỏ lỡ), r (read/snapshot ban đầu). ts_ms là mốc thời gian nguồn, dùng để sắp thứ tự.

Cấu hình connector Debezium (minh hoạ — KHÔNG chạy được, cú pháp Kafka Connect):

{
  "name": "oracle-corebanking-cdc",
  "config": {
    "connector.class": "io.debezium.connector.oracle.OracleConnector",
    "database.hostname": "corebank-db",
    "database.dbname": "ORCL",
    "table.include.list": "BANKING.ACCOUNTS,BANKING.TRANSACTIONS",
    "log.mining.strategy": "online_catalog",
    "tombstones.on.delete": "true"
  }
}

Idempotent & exactly-once ở tầng nạp

Đây là yêu cầu số một của ingestion đáng tin: chạy lại một mẻ (do retry, do lỗi giữa chừng, do replay) phải KHÔNG nhân đôi dữ liệu. Streaming và batch đều có thể giao ít nhất một lần (at-least-once) — nghĩa là một sự kiện có thể tới nhiều lần. Tầng nạp phải tự bảo vệ để hiệu ứng cuối cùng là đúng một lần (exactly-once về mặt kết quả).

Ba kỹ thuật kết hợp:

1. Khoá idempotent (idempotency key): mỗi bản ghi/mẻ có một khoá xác định (primary key + version, hoặc offset Kafka). Nạp cùng khoá hai lần → lần sau ghi đè, không thêm mới.
2. Merge/upsert thay vì insert: dùng MERGE/INSERT ... ON CONFLICT theo primary key. Bản ghi đã có thì cập nhật, chưa có thì thêm — chạy lại luôn cho cùng trạng thái.
3. Dedup theo primary key + version/op: một mẻ có thể chứa nhiều thay đổi của cùng một khoá (dòng 42 update 3 lần). Trước khi merge, phải gộp về bản mới nhất mỗi khoá — dựa vào ts_ms/created_at (order) và op.

Out-of-order là biến chứng: sự kiện có thể tới không đúng thứ tự (do phân vùng Kafka, retry). Nếu áp update cũ đè lên update mới thì hỏng. Phòng: merge có điều kiện version — chỉ ghi đè khi bản đến mới hơn bản đang có (so ts_ms/version), bỏ qua nếu cũ hơn.

SQL 1 — Incremental theo watermark (chạy được)

Lấy các giao dịch phát sinh sau mốc watermark của lần nạp trước (ở đây minh hoạ mốc 2024-01-01) — đúng logic incremental thuần:

-- ▶ Chạy được
SELECT id, account_id, amount, kind, created_at
FROM transactions
WHERE created_at > TIMESTAMP '2024-01-01 00:00:00'
ORDER BY created_at
LIMIT 100;

Trong pipeline thật, hằng số ngày sẽ được thay bằng watermark lưu ở bảng metadata; và nếu chống late data, ta lùi mốc (ví dụ created_at > watermark - INTERVAL '3 days') rồi để bước dedup lo phần trùng lặp.

SQL 2 — Dedup lấy bản mới nhất mỗi khoá (chạy được)

Khi một mẻ chứa nhiều bản ghi cùng khoá, ta chỉ giữ bản mới nhất mỗi account_id bằng ROW_NUMBER(). Đây chính là bước "gộp về bản mới nhất" trước khi MERGE:

-- ▶ Chạy được
WITH ranked AS (
  SELECT
    id, account_id, amount, kind, created_at,
    ROW_NUMBER() OVER (
      PARTITION BY account_id
      ORDER BY created_at DESC
    ) AS rn
  FROM transactions
)
SELECT id, account_id, amount, kind, created_at
FROM ranked
WHERE rn = 1
ORDER BY account_id;

Trong CDC thật, PARTITION BY sẽ theo primary key của bảng nguồn và ORDER BY theo ts_ms DESC, op (delete thắng khi cùng mốc). Giữ rn = 1 cho ta đúng một dòng "trạng thái cuối" mỗi khoá — sẵn sàng cho MERGE idempotent.

MERGE vào bảng đích: SCD1 vs SCD2

Sau dedup, ta đưa "trạng thái cuối mỗi khoá" vào bảng đích. Hai chiến lược:

• SCD Type 1 — ghi đè: chỉ giữ giá trị hiện tại. Update ở nguồn thì đích cũng đổi, mất lịch sử. Hợp cho dữ liệu tham chiếu ít cần truy vết (ví dụ số điện thoại khách).
• SCD Type 2 — giữ lịch sử: mỗi thay đổi tạo một dòng mới với valid_from/valid_to và cờ is_current. Cho phép trả lời "số dư tài khoản này tại thời điểm T là bao nhiêu" — bắt buộc cho đối soát & audit ngân hàng. dbt hỗ trợ SCD2 qua snapshots (sources & snapshots).

Pseudocode MERGE SCD1 (minh hoạ — KHÔNG chạy được, dialect có thể khác):

MERGE INTO dim_accounts AS t
USING staged_latest AS s
ON t.account_id = s.account_id
WHEN MATCHED AND s.ts_ms > t.ts_ms THEN
  UPDATE SET balance = s.balance, ts_ms = s.ts_ms
WHEN MATCHED AND s.op = 'd' THEN
  DELETE
WHEN NOT MATCHED THEN
  INSERT (account_id, balance, ts_ms) VALUES (s.account_id, s.balance, s.ts_ms);

Điều kiện s.ts_ms > t.ts_ms chính là chốt chặn out-of-order và làm phép merge idempotent: chạy lại cùng mẻ không đổi kết quả vì bản đã áp không còn "mới hơn".

Schema drift & evolution ở nguồn

Hệ nguồn đổi schema liên tục: thêm cột, đổi kiểu, đổi tên. Nếu pipeline giả định schema cứng, một cột mới ở Oracle có thể làm gãy cả job hoặc âm thầm mất dữ liệu. Xử lý:

• Phát hiện: so schema mẻ mới với schema đã biết; cột lạ/thiếu → cảnh báo, không âm thầm bỏ.
• Schema registry: đăng ký schema (Avro/Protobuf) qua Kafka Schema Registry, ép quy tắc tương thích (backward/forward). Producer đổi schema vi phạm quy tắc thì bị chặn ngay.
• Chiến lược evolution: cột thêm → thường an toàn, cho phép tự nới bảng đích (nullable). Cột đổi kiểu/xoá → cần review thủ công, không tự áp.

Gắn kết với ingestion metadata-driven: schema của mỗi nguồn khai báo trong metadata, và cấu hình EL (bảng nào, cột mốc nào, chiến lược SCD nào) cũng là khai báo config chứ không hard-code. Thêm bảng mới = thêm một dòng config.

Backfill & reprocess an toàn

Backfill = nạp lại lịch sử (khi thêm cột dẫn xuất mới, khi phát hiện bug ở transform, khi onboard bảng mới). Rủi ro: nạp lại nhân đôi dữ liệu nếu tầng nạp không idempotent. Vì đã dùng MERGE theo khoá, backfill trở nên an toàn: phát lại từ Kafka (hoặc đọc lại raw zone), chạy lại dedup + merge → kết quả hội tụ về đúng một trạng thái. Đây là lý do raw zone bất biến quan trọng: nó là nguồn sự thật để tái xử lý bất kỳ lúc nào, không phụ thuộc DB nguồn còn giữ dữ liệu hay không.

Mẹo backfill: chạy trên cửa sổ có giới hạn (theo ngày/tháng) để kiểm soát tải và có thể tạm dừng; và tách môi trường để không đụng bảng production đang phục vụ (WAP & môi trường).

Chất lượng & contract ngay ở ingest

Đừng đợi tới tầng transform mới kiểm chất lượng. Áp data contract ngay ở cửa nạp (quality & contracts): kiểu dữ liệu, not-null cho khoá, miền giá trị hợp lệ. Bản ghi vi phạm không được làm gãy cả mẻ — thay vào đó đẩy vào quarantine (một topic/bảng riêng) để điều tra, còn phần tốt vẫn chảy tiếp. Cách này giữ pipeline tự phục hồi: một dòng lỗi từ nguồn không đánh sập toàn bộ ingestion.

Công cụ EL

Nguyên tắc chọn: nguồn phổ biến (Postgres, MySQL, Salesforce) → dùng connector có sẵn cho nhanh; core banking Oracle với yêu cầu latency & audit ngặt → thường tự chủ bằng Debezium/GoldenGate để kiểm soát hoàn toàn.

Landing/raw zone bất biến

Nguyên tắc vàng: giữ nguyên bản gốc. Raw zone là append-only, bất biến — mọi sự kiện CDC/mọi mẻ full/incremental được ghi y như nhận được, không sửa, không dedup. Lợi ích: (1) tái xử lý bất kỳ lúc nào nếu logic transform đổi hoặc phát hiện bug; (2) đối soát/audit — khi thanh tra hỏi "dòng này từ đâu ra", raw zone là bằng chứng gốc bất biến. Mọi biến đổi (dedup, merge, SCD) diễn ra sau raw, ở các zone kế tiếp (staging → mart). Raw không bao giờ bị ghi đè.

Use case thực tế

Bài toán: đưa bảng TRANSACTIONS và ACCOUNTS từ core banking Oracle vào lakehouse near-real-time (độ trễ mục tiêu < 5 phút), phục vụ giám sát giao dịch và đối soát cuối ngày. Yêu cầu: bắt được xoá/sửa (khách huỷ giao dịch, điều chỉnh số dư), idempotent (retry không nhân đôi), và giữ raw để audit.

Triển khai:

1. CDC: Debezium đọc redo log Oracle, đẩy vào Kafka topic bank.transactions, bank.accounts. Tải thêm lên Oracle gần như 0 (chỉ đọc log). Delete bắt qua op = 'd'.
2. Raw zone: consumer ghi mọi sự kiện thô (kèm op, ts_ms, offset) vào bảng append-only phân vùng theo ngày — bất biến.
3. Dedup: mỗi 2 phút, gom sự kiện mới, dùng ROW_NUMBER() OVER (PARTITION BY pk ORDER BY ts_ms DESC) giữ trạng thái cuối mỗi khoá (như SQL 2 ở trên).
4. MERGE: upsert vào dim_accounts (SCD1 cho số dư hiện tại) và snapshot SCD2 cho lịch sử số dư — điều kiện s.ts_ms > t.ts_ms chống out-of-order.
5. Quality gate: giao dịch amount null hoặc account_id không tồn tại → quarantine, cảnh báo, phần còn lại vẫn nạp.

Kết quả: độ trễ nguồn→lakehouse ~3 phút; một sự cố consumer restart giữa mẻ được retry — nhờ MERGE idempotent, không sinh giao dịch trùng. Cuối tháng, thanh tra yêu cầu tái dựng số dư một tài khoản tại ngày 15: đọc raw zone + SCD2, dựng lại chính xác không cần đụng Oracle.

Ghi nhớ

• Ingestion là chữ E-L đứng trước T của dbt; phần lớn sự cố dữ liệu bắt nguồn từ đây.
• Ba mẫu: full load (đơn giản, tốn), incremental theo watermark (rẻ, nhưng bẫy late data + không bắt DELETE), CDC (near-real-time, bắt cả xoá/sửa).
• CDC log-based (Debezium/GoldenGate) là chuẩn cho core banking: gần như không tải nguồn, bắt DELETE qua op = 'd'.
• Idempotent = khoá + MERGE/upsert theo primary key + dedup (ROW_NUMBER theo version) + điều kiện version chống out-of-order.
• SCD1 ghi đè (giá trị hiện tại), SCD2 giữ lịch sử (đối soát/audit) — dbt snapshots hỗ trợ SCD2.
• Xử lý schema drift bằng phát hiện + schema registry + quy tắc tương thích; khai báo EL bằng config metadata-driven.
• Raw zone bất biến (append-only) là nguồn sự thật cho backfill và audit — mọi biến đổi diễn ra sau raw.
• Áp contract & quarantine ngay ở cửa nạp để pipeline tự phục hồi trước dòng lỗi.