DataOps nâng cao 5 — Data Quality & Data Contracts thực chiến

Chất lượng không phải "chạy vài test cho vui"

Bài DataOps 4 — testing & quality và Governance 3 — data quality đã đặt nền: sáu chiều chất lượng, khái niệm test. Bài này đi sâu một bậc — thực chiến: chọn công cụ nào cho lớp nào, đặt test ở đúng chốt trong pipeline, và biến "kỳ vọng chất lượng" thành hợp đồng dữ liệu (data contract) có hiệu lực trong CI lẫn runtime.

Trong ngân hàng, chất lượng dữ liệu không phải là chuyện "đẹp bảng dashboard". Một cột city bị null làm hỏng báo cáo phân bố địa lý; một account_no trùng làm double-count dư nợ; một giao dịch mồ côi (orphan) không khớp tài khoản làm lệch số liệu đối soát gửi NHNN. Chốt chặn phải nằm trước báo cáo tuân thủ, không phải sau khi kiểm toán phát hiện. Đó là tinh thần xuyên suốt bài này.

Ba lớp công cụ test — dùng cái nào, ở đâu

Không có một công cụ "trị bách bệnh". Ba công cụ phổ biến nhất phục vụ ba lớp khác nhau; hiểu ranh giới của chúng quan trọng hơn học cú pháp.

Nguyên tắc chọn:

• dbt tests — mặc định cho mọi thứ đã ở trong warehouse và gắn với logic transform. Test đi cùng model, versioned cùng Git, chạy trong CI (xem DataOps nâng cao 4 — CI/CD). Đây là 80% nhu cầu.
• Great Expectations — khi cần kiểm ở ranh giới ingest, trước cả khi dbt chạm tới: file CSV/Parquet từ core banking, payload API từ đối tác, bảng staging từ CDC. GE validate được cả những thứ dbt chưa "nhìn thấy".
• Soda — khi cần quét theo lịch độc lập DAG: một job cron mỗi giờ kiểm freshness và volume trên nhiều nguồn, gửi alert Slack, không cần dbt run.

Trong thực tế NCB điển hình: GE ở ingest → dbt tests trong transform → Soda quét monitoring nền. Ba lớp bổ sung nhau, không thay thế.

dbt tests: từ generic tới custom

dbt cung cấp bốn generic test dùng khai báo trong YAML — đây là xương sống:

# models/marts/schema.yml — YAML, không đánh dấu chạy được
models:
  - name: dim_accounts
    columns:
      - name: account_no
        tests: [not_null, unique]
      - name: currency
        tests:
          - accepted_values:
              values: ['VND', 'USD', 'EUR']
      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: id

Khi generic không đủ, có ba nấc leo thang:

1. Singular test — một file .sql trong tests/, là một SELECT trả về các dòng vi phạm. Có dòng nào ⇒ test fail. Đây chính là dạng ta chạy được trên sandbox bên dưới.
2. Custom generic test — đóng gói một singular test thành macro tái dùng cho nhiều model (ví dụ assert_positive_balance).
3. Packages — dbt-utils (test equal_rowcount, expression_is_true, recency) và dbt-expectations (port GE sang dbt: expect_column_values_to_be_between, expect_column_mean_to_be_between...). Cài qua packages.yml, khỏi viết lại.

Đặt test ĐÚNG VỊ TRÍ

Test đặt sai chỗ thì hoặc bắt lỗi quá muộn, hoặc không bắt được. Ba chốt bắt buộc:

Chốt 1 — source freshness ở đầu vào. Trước khi transform, kiểm dữ liệu có mới không. dbt có source freshness (khai loaded_at_field + ngưỡng warn_after/error_after); GE kiểm schema và null ở file thô. Ngân hàng: nếu bảng giao dịch chưa cập nhật đến cuối ngày T-1, dừng ngay, đừng chạy báo cáo trên dữ liệu cũ.

Chốt 2 — test sau transform. Sau khi model build, chạy not_null, unique, relationships, accepted_values trên bảng kết quả. Bắt lỗi logic transform: join nhân đôi dòng, CASE sót nhánh, khoá ngoại gãy.

Chốt 3 — circuit breaker trước publish. Đây là chốt quan trọng nhất và hay bị bỏ sót. Ngay trước khi ghi ra mart/báo cáo, chạy các kiểm tra đối soát (reconciliation): tổng kiểm soát (control total) khớp nguồn không? Nếu fail thì CHẶN publish — dữ liệu cũ ở mart vẫn còn tốt hơn số liệu sai đè lên. Circuit breaker biến "phát hiện sau" thành "chặn trước".

Năm data test chạy được trên sandbox

Mỗi block dưới là một singular test kiểu dbt / GE-style: một SELECT trả về các dòng vi phạm (hoặc chỉ số đối soát). Chạy được trực tiếp trên sandbox PostgreSQL read-only.

(a) not_null — customers.city không được rỗng. Có dòng trả về ⇒ vi phạm:

-- ▶ Chạy được
SELECT id, full_name
FROM customers
WHERE city IS NULL;

(b) uniqueness — accounts.account_no phải duy nhất. So COUNT với COUNT DISTINCT; nếu lệch là có trùng. Truy vấn liệt kê thẳng các account_no xuất hiện >1 lần:

-- ▶ Chạy được
SELECT account_no, COUNT(*) AS so_lan
FROM accounts
GROUP BY account_no
HAVING COUNT(*) > 1;

(c) relationship / orphan — giao dịch mồ côi không khớp tài khoản. LEFT JOIN rồi lọc accounts.id IS NULL: mỗi dòng là một giao dịch có account_id không tồn tại trong accounts:

-- ▶ Chạy được
SELECT t.id, t.account_id, t.amount
FROM transactions t
LEFT JOIN accounts a ON t.account_id = a.id
WHERE a.id IS NULL;

(d) accepted range — số dư âm hoặc giao dịch giá trị 0. Bất biến nghiệp vụ: balance không được âm và amount không được bằng 0. Dòng trả về là vi phạm cần điều tra:

-- ▶ Chạy được
SELECT a.account_no, a.currency, a.balance
FROM accounts a
WHERE a.balance < 0;

(e) reconciliation control total — tổng kiểm soát theo loại tiền. Chốt circuit breaker: tổng số dư và số lượng tài khoản theo từng currency. Đội báo cáo đối chiếu con số này với tổng từ nguồn; lệch ⇒ chặn publish:

-- ▶ Chạy được
SELECT currency, COUNT(*) AS so_tk, SUM(balance) AS tong_du
FROM accounts
GROUP BY currency
ORDER BY tong_du DESC;

Năm truy vấn này chính là những gì bạn gói vào tests/*.sql (dbt singular test) hoặc chuyển thành expectation trong GE. Chúng khai báo kỳ vọng biết trước về dữ liệu — hạt nhân của data test.

Data Contracts thực chiến

Test là kỳ vọng của đội tiêu thụ. Data contract đi xa hơn: là cam kết chính thức của đội sản xuất (nguồn) với đội tiêu thụ về hình dạng và chất lượng dữ liệu. Nền lý thuyết ở Data Mesh 4 — data contracts; ở đây ta viết và enforce nó.

Một contract mô tả bằng YAML gồm: schema (cột + kiểu), ràng buộc (not null, unique, range), SLO (freshness, độ trễ), ownership (ai chịu trách nhiệm), và phân loại PII:

# contracts/accounts_v1.yaml — YAML, không đánh dấu chạy được
dataset: core.accounts
version: 1
owner: team-core-banking          # đội NGUỒN chịu trách nhiệm
consumers: [team-regulatory-reporting]
slo:
  freshness: 24h                  # cập nhật ít nhất mỗi 24 giờ
  availability: 99.5%
schema:
  - name: account_no
    type: string
    constraints: [not_null, unique]
    pii: false
  - name: customer_id
    type: integer
    constraints: [not_null]
    references: core.customers.id # ràng buộc quan hệ
  - name: balance
    type: numeric
    constraints: ["min: 0"]       # không âm
  - name: currency
    type: string
    constraints: ["accepted: [VND, USD, EUR]"]

Enforcement: CI và runtime — hai lớp

Contract chỉ có giá trị khi được thực thi tự động, không phải tài liệu để trong wiki.

Trong CI — chặn merge nếu phá vỡ contract. Khi đội nguồn sửa schema, một job CI so schema mới với contract đã đăng ký. Xoá cột currency mà consumer đang dùng, đổi kiểu balance từ numeric sang string ⇒ CI fail, chặn merge. Cùng cơ chế cổng chặn ở DataOps nâng cao 4:

# .github/workflows/contract-check.yml — YAML, không đánh dấu chạy được
- name: Validate data contract
  run: datacontract test contracts/accounts_v1.yaml --schema warehouse
  # exit code != 0 → chặn merge

Runtime — validate lúc ingest. Khi dữ liệu thực chảy qua, GE checkpoint hoặc Soda scan kiểm nó có tuân contract không (freshness, range, referential). Vi phạm ⇒ chuyển vào quarantine (mục sau), không để lan xuống.

Schema evolution & registry

Contract sống nhiều năm, schema phải tiến hoá tương thích ngược (backward-compatible): được thêm cột optional, không được xoá/đổi kiểu cột consumer đang dùng. Thay đổi phá vỡ ⇒ bump version (accounts_v2), chạy song song hai version tới khi consumer chuyển hết.

Với streaming, contract sống trong schema registry (ví dụ Confluent Schema Registry cho Avro/Protobuf). Producer đăng ký schema; registry từ chối schema không tương thích ngược ngay tại thời điểm publish message — enforcement tự động, không cần CI.

Test vs Anomaly detection — bổ sung, không thay thế

Đây là phân biệt hay bị lẫn:

Test bắt "cái ta biết cần kiểm". Anomaly bắt "cái ta không ngờ": một job upstream lặng lẽ chết làm số dòng tụt, một đổi mã tiền tệ làm phân bố lệch. Hai cái bổ sung nhau — dùng test cho bất biến nghiệp vụ cứng, dùng anomaly cho volume/freshness/phân bố. Chi tiết observability ở DataOps nâng cao 6.

Xử lý dữ liệu xấu: quarantine, không xoá

Khi test/contract fail, tuyệt đối không để dữ liệu xấu chảy tiếp xuống báo cáo. Chiến lược:

• Quarantine (bảng cách ly) — tách các dòng vi phạm ra bảng _quarantine riêng thay vì drop. Dòng tốt vẫn chảy, dòng xấu giữ lại để điều tra (không mất bằng chứng). Ví dụ: giao dịch orphan (test c) → transactions_quarantine.
• Dead-letter — trong streaming, message không tuân contract đi vào dead-letter topic thay vì làm nghẽn stream chính.
• Cảnh báo — mỗi lần quarantine phát alert cho owner (theo owner trong contract), kèm số dòng và lý do.
• Không để lan xuống — circuit breaker ở chốt 3 đảm bảo một mart có dữ liệu bị nghi ngờ thì giữ bản cũ, không publish bản mới sai.

Tinh thần: dữ liệu xấu bị cô lập và báo động, không âm thầm bị bỏ qua cũng không âm thầm phá báo cáo.

Đo lường chất lượng

Không đo thì không quản được. Ba chỉ số cốt lõi:

• DQ score — điểm tổng hợp theo dataset (ví dụ: trọng số các chiều completeness/validity/uniqueness). Đưa lên dashboard, theo dõi xu hướng.
• % test pass — tỉ lệ test xanh trên tổng test mỗi lần chạy. Tụt đột ngột = tín hiệu sớm.
• Time-to-detect — thời gian từ lúc lỗi phát sinh tới lúc phát hiện. Chốt càng sớm (chốt 1 thay vì để kiểm toán tìm ra), con số này càng nhỏ. Đây là thước đo thật sự của hệ thống DQ.

Use case thực tế

Bối cảnh. Đội regulatory-reporting NCB dựng báo cáo phân loại nợ hàng ngày gửi NHNN, lấy accounts và transactions từ domain core-banking. Tháng trước, một lần đổi schema bên nguồn khiến vài account_no bị nhân đôi trong quá trình migrate CDC, làm dư nợ nhóm 2 bị double-count +6,2% — chỉ phát hiện khi NHNN hỏi ngược. Time-to-detect: 9 ngày. Không chấp nhận được.

Giải pháp — ba chốt + contract.

1. Contract accounts_v1.yaml: account_no not_null + unique, balance min:0, currency accepted [VND,USD,EUR], customer_id references customers.id, SLO freshness 24h, owner = core-banking. Enforce trong CI của repo core-banking: mọi PR đổi schema accounts phải qua datacontract test.
2. Chốt 1 (ingest, GE): kiểm freshness loaded_at ≤ 24h và null account_no trên bảng staging trước khi dbt chạm.
3. Chốt 2 (sau transform, dbt): năm singular test ở trên — trong đó test (b) uniqueness account_no chính là cái đã bắt được lỗi double-count. Dòng trùng → accounts_quarantine, alert owner.
4. Chốt 3 (circuit breaker): trước khi publish mart báo cáo, chạy reconciliation (test e) — SUM(balance) theo currency phải khớp control total từ nguồn trong ngưỡng ±0,01%. Lệch ⇒ chặn publish, giữ báo cáo hôm trước.

Kết quả (minh hoạ, mục tiêu vận hành). Lỗi cùng loại lần sau bị bắt ngay trong CI/transform trong ngày thay vì 9 ngày. Time-to-detect: 9 ngày → dưới 1 giờ. DQ score dataset accounts theo dõi công khai; % test pass là điều kiện bắt buộc để job publish chạy. Số liệu sai không còn cơ hội vào báo cáo tuân thủ.

Ghi nhớ

• Ba lớp công cụ, ba chốt: dbt tests (trong transform, mặc định 80%), Great Expectations (ranh giới ingest/nguồn ngoài), Soda (quét theo lịch). Chọn theo vị trí cần kiểm, không theo sở thích.
• Đặt test đúng chốt: source freshness đầu vào → test sau transform → circuit breaker trước publish. Chốt 3 chặn dữ liệu xấu, giữ bản cũ còn hơn đè bản sai.
• Data contract = cam kết của đội nguồn, viết YAML (schema, kiểu, ràng buộc, SLO, owner, PII), enforce cả trong CI (chặn merge) và runtime (validate ingest); streaming dùng schema registry.
• Schema evolution tương thích ngược: chỉ thêm cột optional; thay đổi phá vỡ ⇒ bump version, chạy song song tới khi consumer chuyển hết.
• Test và anomaly bổ sung nhau: test bắt lỗi đã lường trước (bất biến cứng), anomaly bắt lỗi chưa biết (volume/freshness/phân bố).
• Dữ liệu xấu: quarantine, không xoá. Tách bảng lỗi / dead-letter, alert owner, không để lan xuống báo cáo tuân thủ.
• Đo để quản: DQ score, % test pass, và quan trọng nhất là time-to-detect — chốt càng sớm, phát hiện càng nhanh.
• Năm block SQL trong bài là singular test thật (not_null, uniqueness, orphan, range, reconciliation) — gói thẳng vào tests/*.sql của dbt hoặc chuyển sang expectation GE.