Kiểm thử chất lượng và Mạch dừng dữ liệu (Data Testing & Circuit Breaker)

Trong phát triển phần mềm truyền thống (Software Engineering), Unit Test giúp bạn đảm bảo logic code hoạt động đúng đắn trước khi deploy. Tuy nhiên, trong Kỹ thuật Dữ liệu (Data Engineering), Code chạy đúng là chưa đủ, vì bản chất của Dữ liệu (Data) là liên tục biến động và đầy rẫy sự hỗn loạn.

Một ngày đẹp trời, hệ thống CRM của đối tác đột ngột trả về chuỗi \$10.5 thay vì kiểu số thực 10.5, hoặc team Backend vô tình gỡ bỏ cột user_id trong Kafka event. Data Pipeline của bạn (Code) vẫn chạy trơn tru không báo lỗi, nhưng Dữ liệu đầu ra đã biến thành “Rác” (Garbage In - Garbage Out), làm sai lệch toàn bộ Dashboard doanh thu của CEO.

Bài viết này bỏ qua những định nghĩa cơ bản để đi sâu vào cách thiết kế hệ thống kiểm thử dữ liệu (Data Testing) dưới góc nhìn Kiến trúc Hệ thống (Architecture). Chúng ta sẽ phân tích mô hình Circuit Breaker, kiến trúc Write-Audit-Publish (WAP) (do Netflix tiên phong), và những sự đánh đổi (Trade-offs) khốc liệt về tài nguyên (Compute Cost) khi thực thi trên quy mô Petabyte.

---

1. Kiến trúc Thực thi Vật lý (Physical Execution Patterns)

Khi nào và ở đâu chúng ta nên kích hoạt việc chạy Test? Thay vì cắm test bừa bãi ở mọi nơi, các Data Platform hiện đại áp dụng các Design Pattern cực kỳ cụ thể để kiểm soát luồng dữ liệu.

1.1. Circuit Breaker (Mạch dừng)

Mượn khái niệm từ kiến trúc Microservices (Ví dụ: thư viện Hystrix của Netflix), Data Circuit Breaker là cơ chế ngắt mạch toàn bộ hoặc một phần Pipeline khi dữ liệu không đạt chuẩn. Nếu tỷ lệ lỗi vượt quá ngưỡng cho phép (Error SLA), Pipeline sẽ bị dừng ngay lập tức (Fail-fast) thay vì tiếp tục nạp dữ liệu sai vào hệ thống phục vụ (Serving Layer).

Code Thực chiến (dbt YAML): Dưới đây là ví dụ cấu hình dbt test với cơ chế ngắt mạch phân cấp warn và error.

models:
  - name: fct_transactions
    columns:
      - name: transaction_amount
        tests:
          - not_null
          # Dùng dbt_expectations để test miền giá trị
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 1000000
              config:
                # SEV-3: Chỉ bắn cảnh báo (Alert), KHÔNG ngắt mạch
                severity: warn
      - name: user_id
        tests:
          - not_null:
              config:
                # SEV-1: Ngắt mạch lập tức (Circuit Breaker) nếu vi phạm
                severity: error
          - relationships:
              to: ref('dim_users')
              field: id

Nếu test được đánh dấu error bị fail, dbt sẽ trả về exit code lỗi. Trình điều phối (như Airflow) sẽ bắt được lỗi này và ngừng trigger các task downstream, bảo vệ thành công các bảng Aggregate và Dashboard phía sau.

1.2. Write-Audit-Publish (WAP) Pattern

Circuit Breaker rất tốt, nhưng nó có một điểm yếu chết người: Dữ liệu bẩn đã bị ghi một phần vào Production Table trước khi Pipeline bị ngắt, gây ra tình trạng “nửa nạc nửa mỡ” (Partial Write).

Để giải quyết triệt để, Netflix và Airbnb (với Midas framework) đã thiết kế ra Write-Audit-Publish (WAP). WAP giải quyết bài toán: “Làm sao để test dữ liệu thực tế mà không làm ảnh hưởng đến người dùng cuối?”

1. Write (Ghi): Spark/Flink ghi dữ liệu mới vào một nhánh ẩn (Staging / Branch) hoặc một phân vùng tạm (Temporary Partition) chưa được phơi bày.
2. Audit (Kiểm toán): Khởi động các Data Tests (Great Expectations, Soda, dbt) quét trực tiếp trên nhánh ẩn đó.
3. Publish (Phát hành): Nếu Audit Pass (Xanh), thực hiện thao tác Hoán đổi Nguyên tử (Atomic Swap) để phơi bày dữ liệu ra Main branch cho người dùng. Nếu Fail (Đỏ), nhánh ẩn bị vứt bỏ, Production Table không hề bị “vấy bẩn”.

Mô phỏng WAP Pattern với Apache Iceberg & Trino: Nhờ tính năng Time Travel và Branching của Iceberg, WAP trở nên cực kỳ dễ dàng bằng vài câu SQL.

-- 1. WRITE: Tạo một nhánh rẽ (branch) ẩn có tên 'audit_branch'
ALTER TABLE prod.lakehouse.fct_sales
CREATE BRANCH audit_branch;

-- Ingest data vào nhánh ẩn (Data consumer không nhìn thấy)
INSERT INTO prod.lakehouse.fct_sales FOR VERSION AS OF 'audit_branch'
SELECT * FROM raw.sales_kafka_dump;

-- 2. AUDIT: Chạy câu query kiểm tra dữ liệu trên nhánh 'audit_branch'
-- Nếu kết quả trả về > 0, tức là có lỗi Null, Airflow sẽ báo Fail.
SELECT COUNT(*)
FROM prod.lakehouse.fct_sales FOR VERSION AS OF 'audit_branch'
WHERE transaction_id IS NULL;

-- 3. PUBLISH: Nếu Audit Pass, fast-forward nhánh main lên audit_branch (Atomic)
CALL catalog.system.fast_forward('prod.lakehouse.fct_sales', 'main', 'audit_branch');

---

2. Rủi ro Vận hành & Systemic Trade-offs

Việc cắm hàng trăm cái Test vào Pipeline không phải là “bữa trưa miễn phí”. Dưới đây là những cái giá phải trả và cách các Staff Engineer xử lý.

🚨 Rủi ro 1: Compute Cost vs. Pipeline Latency (Phồng rộp Chi phí)

Vấn đề: Các bài test phức tạp (đặc biệt là test relationships (Khóa ngoại) hay test trùng lặp unique trên các bảng Fact có hàng chục tỷ dòng) đòi hỏi các thao tác Network Shuffle khổng lồ. Hậu quả là tiền Cloud (Snowflake Credit, BigQuery Bytes Billed) tăng vọt theo cấp số nhân và làm chậm SLA của pipeline hàng tiếng đồng hồ. Lỗi Cartesian Explosion (Nổ tổ hợp) rất dễ xảy ra nếu câu lệnh test SQL không có điều kiện WHERE để giới hạn phân vùng.

✅ Giải pháp (Systemic Trade-off):

• Đánh đổi tính chính xác tuyệt đối lấy hiệu năng bằng cách Lấy mẫu (Sampling).
• Chỉ chạy test trên phân vùng dữ liệu mới nhất (Incremental Testing) thay vì quét toàn bộ bảng (Full Table Scan).

Code Thực chiến (dbt Incremental Test tối ưu chi phí): Thay vì test tính Unique trên 10 tỷ dòng, ta chỉ test trên những dòng được chèn trong 3 ngày gần nhất.

-- tests/assert_recent_transactions_unique.sql
-- Chi phí quét giảm từ 100$ xuống còn 0.5$
SELECT transaction_id, count(*)
FROM {{ ref('fct_transactions') }}
-- Chỉ quét phân vùng dữ liệu mới (Partition Pruning)
WHERE created_at >= CURRENT_DATE - INTERVAL '3 days'
GROUP BY transaction_id
HAVING count(*) > 1

🚨 Rủi ro 2: Mệt mỏi vì Cảnh báo (Alert Fatigue)

Vấn đề: Khi bạn thiết lập test với ngưỡng tĩnh (Static Thresholds), ví dụ: row_count > 100,000 thì pass. Vào dịp Black Friday, lượng đơn hàng tăng vọt lên 1,000,000 dòng. Hoặc vào Mùng 1 Tết, lượng đơn rớt thảm hại còn 10,000 dòng. Cả hai trường hợp đều khiến dbt báo lỗi FAIL và kích hoạt PagerDuty (False Positives). Kỹ sư bị “spam” cảnh báo giữa đêm, dẫn đến trạng thái Alert Fatigue - mệt mỏi, tức giận và bắt đầu phớt lờ cảnh báo.

✅ Giải pháp từ Uber DQM (Data Quality Monitor): Chuyển từ Static Threshold sang Statistical Modeling (Mô hình thống kê) và Anomaly Detection (Phát hiện bất thường). Hệ thống sẽ học từ dữ liệu lịch sử để tự động điều chỉnh dải băng tin cậy (Confidence Bands) dựa trên tính mùa vụ (Seasonality).

# Pseudo-code minh họa Data Observability theo kiểu Uber
from scipy.stats import zscore
import pandas as pd

def detect_volume_anomaly(daily_volume_series: pd.Series, threshold_z=3.0):
    """
    Thay vì fix cứng số dòng, dùng Z-Score để tìm bất thường
    dựa trên phân phối chuẩn của 30 ngày gần nhất.
    """
    # Tính điểm Z-Score
    z_scores = zscore[daily_volume_series]
    latest_z = z_scores.iloc[-1]

    # Nếu dữ liệu lệch quá 3 độ lệch chuẩn (3 Sigma]
    if abs(latest_z) > threshold_z:
        trigger_pagerduty_alert(f"Volume anomaly detected! Z-Score: {latest_z}")
    else:
        log_info("Volume is within historical norms. (Maybe just Black Friday effect)")

🚨 Rủi ro 3: OOMKilled trên Test Runner [Tràn RAM]

Vấn đề: Khi dùng thư viện Python (như Pandas hoặc PandasProfiler) chạy trên Kubernetes Pod (Worker) để kéo (pull) hàng triệu dòng dữ liệu từ Data Warehouse về RAM để test. Pod sẽ lập tức hết sạch Memory, HĐH bắn tín hiệu tự vệ, gây ra lỗi OOMKilled (Exit Code 137).

✅ Giải pháp (Push-down Compute): Tuyệt đối không di chuyển dữ liệu (Data Movement) về nơi tính toán. Hãy đẩy logic tính toán (Compute) xuống nơi chứa dữ liệu. Sử dụng dbt hoặc Great Expectations với SqlAlchemyExecutionEngine để compile luật Test thành các câu lệnh SQL Native (COUNT, SUM, GROUP BY) và để Warehouse tự xử lý dưới dạng Distributed Query.

---

3. Vòng Đời Dữ Liệu Khuyết Tật (Data Quarantine)

Khi Circuit Breaker ngắt mạch, dữ liệu lỗi sẽ đi đâu? Vứt bỏ đi là một thảm họa vì bạn sẽ mất doanh thu.

Các hệ thống Enterprise sẽ định tuyến các record lỗi vào Quarantine Table (Bảng cách ly) hoặc Dead Letter Queue (DLQ). Tại đây, Data Engineer sẽ phân tích nguyên nhân gốc rễ. Sau khi kịch bản lỗi (ví dụ: đối tác sửa lại định dạng API) được khắc phục, các record trong Bảng cách ly sẽ được chạy lại qua Pipeline (Re-processing/Backfilling) để bù đắp dữ liệu bị thiếu, đảm bảo nguyên tắc Zero Data Loss.

---

Nguồn Tham Khảo (References)

1. Uber Engineering: [Monitoring Data Quality at Scale with Statistical Modeling][https://www.uber.com/en-VN/blog/]
2. Netflix Tech Blog: Whoops, the Numbers are Wrong! Scaling Data Quality @ Netflix
3. Apache Iceberg / Project Nessie: [Write-Audit-Publish Pattern for Data Lakes][https://projectnessie.org/features/wap/]
4. Sách: Designing Data-Intensive Applications - Martin Kleppmann (Chương 11: Stream Processing - Cảnh báo và Quality)
5. dbt Best Practices: [Data Testing Configuration & Incremental Tests](https://docs.getdbt.com/docs/build/data-tests]