Testing cho Data PipelineTesting for Data Pipelines

Thuộc bộ kiến thức Data Engineer Roadmap.

Tổng quan

Testing cho application thường chỉ hỏi một câu: code có đúng không? Testing cho data pipeline phải hỏi đồng thời hai câu: code có đúng không, và dữ liệu có đúng không? Một transformation function có thể hoàn toàn không có lỗi logic — mọi nhánh đều được cover, mọi edge case đều được xử lý, mọi test đều xanh — nhưng vẫn cho ra một báo cáo sai, vì join key ở upstream âm thầm đổi cardinality, một hệ thống nguồn bắt đầu trả về NULL thay vì 0 như trước, hoặc một API đổi cách phân trang và một batch job âm thầm bỏ sót trang cuối. Không có cái nào trong số đó là bug ở code cả. Đó là một data bug, và chỉ có thể bị phát hiện bởi những test nhìn vào chính dữ liệu, chứ không chỉ nhìn vào logic xử lý dữ liệu đó.

Đây chính là điểm khác biệt cốt lõi giữa testing data pipeline và testing một backend service thông thường, và bài viết này giả định bạn đã nắm được từ vựng chung của testing — unit test, integration test, functional/E2E test, test double, TDD, CI gating — vốn đã được trình bày chi tiết ở Backend — Testing. Thay vì lặp lại nội dung đó, bài này tập trung vào những gì thực sự khác biệt hoặc khó hơn khi đối tượng được test là một pipeline di chuyển và biến đổi dữ liệu: test transformation logic một cách độc lập (nhắc lại ngắn gọn kèm ví dụ mang màu sắc pipeline), integration test và functional test trên hệ thống nguồn/đích thực tế hoặc mô phỏng, các nhóm data-specific test không có tương đương thật sự trong application testing (schema, not-null/uniqueness/referential-integrity, distribution/anomaly, business-rule), idempotency testing — một mối quan tâm đặc thù của các hệ thống mà việc chạy pipeline làm thay đổi state như một side effect — smoke test và load test cho pipeline, và thách thức thực tế của việc test ở quy mô production mà không đụng vào dữ liệu production.

Bài liên quan: ETL/ELT & Data Pipelines · Data Quality, Governance & Metadata · CI/CD for Data Engineering.

Kiến thức nền tảng

Mối quan tâm kép: code correctness vs. data correctness

Trong một backend web thông thường, input mà test quan tâm là do chính test tự dựng lên — bạn tự tạo một request payload, một mock user, một fixture row, và hoàn toàn kiểm soát nó. Nếu test pass, bạn có niềm tin thực sự rằng code xử lý đúng shape input đó, vì shape đó chính là contract.

Input thực của một data pipeline lại không do ai trong data team dựng lên cả — đó là bất cứ thứ gì hệ thống nguồn đang emit ra ngày hôm nay, và nó có thể drift mà không có bất kỳ thay đổi code nào ở cả hai phía. Một cột NOT NULL trong database nguồn có thể bắt đầu nhận null sau khi phía application ship một bug; một field “status” có thể xuất hiện một giá trị enum mới mà không ai báo trước; một field currency có thể âm thầm chuyển từ cents sang dollars sau khi vendor migrate hệ thống. Code của pipeline không hề thay đổi, các test trên code đó vẫn pass, vậy mà output giờ lại sai. Đây là lý do testing cho data pipeline cần thêm một trục test thứ hai, độc lập, khẳng định những điều về dữ liệu đang chảy qua hệ thống ngay lúc này, chứ không chỉ về logic biến đổi dữ liệu. Hãy ghi nhớ cả hai trục này xuyên suốt bài viết:

Trục	Câu hỏi nó trả lời	Được test ở đâu	Fail khi nào
Code correctness	”Transformation function này có làm đúng như tôi thiết kế không?”	Unit test, chạy trên các cặp input/output cố định	Ai đó sửa logic và làm hỏng một case đã biết
Data correctness	”Dữ liệu đang chảy qua pipeline có thực sự đúng như giả định ban đầu không?”	Data test (schema, null, uniqueness, distribution, business rule), chạy trên dữ liệu thật hoặc gần thật	Thế giới bên ngoài thay đổi — một hệ thống nguồn, một team upstream, hành vi người dùng — dù code không đổi

Một bộ test pipeline trưởng thành phải dành thời gian và CI minutes cho cả hai trục. Những team chỉ làm trục thứ nhất (vì đó là phản xạ quen thuộc từ application testing) sẽ bị bất ngờ bởi những vụ data corruption âm thầm mà không một unit test nào có thể bắt được.

Nhắc lại: unit test một transformation function

Cơ chế hoàn toàn giống với unit testing trong application: cô lập một function, đưa vào input đã biết, assert trên output đã biết, không I/O, không database, không network. Xem Backend — Testing để có đầy đủ pattern AAA (Arrange-Act-Assert) và taxonomy của test double; dưới đây là pattern đó áp dụng cho một transformation nhỏ trong pipeline.

# transforms.py
from datetime import date

def normalize_order(raw: dict) -> dict:
    """Coerce a raw order record into the canonical shape used downstream."""
    if raw.get("amount_cents") is None:
        raise ValueError("amount_cents is required")

    return {
        "order_id": str(raw["order_id"]),
        "amount": round(raw["amount_cents"] / 100, 2),
        "currency": (raw.get("currency") or "USD").upper(),
        "order_date": date.fromisoformat(raw["order_date"]),
        "is_refund": raw.get("amount_cents", 0) < 0,
    }

# test_transforms.py
import pytest
from datetime import date
from transforms import normalize_order

def test_converts_cents_to_decimal_amount():
    raw = {"order_id": 1, "amount_cents": 1999, "currency": "usd", "order_date": "2026-01-05"}
    result = normalize_order(raw)
    assert result["amount"] == 19.99

def test_defaults_missing_currency_to_usd():
    raw = {"order_id": 2, "amount_cents": 500, "order_date": "2026-01-05"}
    result = normalize_order(raw)
    assert result["currency"] == "USD"

def test_flags_negative_amounts_as_refunds():
    raw = {"order_id": 3, "amount_cents": -1000, "order_date": "2026-01-05"}
    result = normalize_order(raw)
    assert result["is_refund"] is True

def test_raises_when_amount_is_missing():
    raw = {"order_id": 4, "order_date": "2026-01-05"}
    with pytest.raises(ValueError):
        normalize_order(raw)

Bộ test này chạy trong vài mili-giây, không cần database, và pin chính xác hành vi của normalize_order. Tuy nhiên nó không nói lên được gì về việc bảng orders thực tế có cột amount_cents đúng kiểu dữ liệu mong đợi hay không, hoặc liệu batch load đêm qua có sinh ra số lượng row hợp lý hay không — đó là nội dung của các phần sau.

Khái niệm chính

Integration testing: hệ thống thật, dữ liệu sandboxed

Một integration test cho pipeline kiểm tra những phần mà unit test về bản chất không thể chạm tới: bước extract có thực sự authenticate đúng với API nguồn và phân trang đúng không, bước load có thực sự ghi row vào warehouse đích với đúng kiểu dữ liệu không, một Kafka consumer có thực sự commit offset sau khi ghi thành công không. Mục tiêu là chạy trên engine thật — một Postgres thật, một Kafka broker thật, một object store tương thích S3 thật — vì mock rất dễ âm thầm sai lệch so với hành vi thật đúng vào những edge case (constraint violation, encoding, pagination cursor) quan trọng nhất.

Cách thực tế để có được “thật nhưng an toàn” chính là pattern Testcontainers quen thuộc từ integration testing của application, cộng thêm sandboxed cloud schema/bucket cho những hệ thống khó container hóa:

# test_pipeline_integration.py
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.kafka import KafkaContainer
from pipeline import extract_new_orders, load_orders

@pytest.fixture(scope="module")
def source_db():
    with PostgresContainer("postgres:16-alpine") as pg:
        conn = pg.get_connection_url()
        # seed a source schema that mimics production shape
        yield conn

@pytest.fixture(scope="module")
def warehouse_db():
    with PostgresContainer("postgres:16-alpine") as pg:
        yield pg.get_connection_url()

def test_pipeline_moves_new_orders_end_to_end(source_db, warehouse_db):
    rows = extract_new_orders(source_db, since="2026-01-01")
    load_orders(warehouse_db, rows)

    # assert against the real destination, not a mock
    import psycopg
    with psycopg.connect(warehouse_db) as conn:
        count = conn.execute("SELECT count(*) FROM orders").fetchone()[0]
        assert count == len(rows)

Có hai nguyên tắc giữ cho những test này đáng tin cậy và nhanh: không bao giờ trỏ chúng vào production credentials hay production schema (dùng container dùng-một-lần hoặc một sandbox project/schema riêng cho mỗi lần chạy test), và truncate/tạo lại state giữa các test để thứ tự chạy không bao giờ ảnh hưởng kết quả — cùng nguyên tắc cô lập đã nêu cho integration test của application trong Backend — Testing.

Functional và end-to-end testing: input đã biết, output kỳ vọng đã biết

Một functional test (hay end-to-end test) cho pipeline chạy toàn bộ pipeline — extract, transform, load — trên một tập dữ liệu input nhỏ, được dựng thủ công, mà bạn đã biết trước, bằng tính toán tay, chính xác output phải là gì. Đây là phiên bản dành cho pipeline của một black-box API test: bạn không quan tâm bên trong hoạt động thế nào, chỉ quan tâm một input cụ thể phải cho ra một output cụ thể, đúng về mặt nghiệp vụ.

Cấu trúc điển hình: check một fixture file (vài chục row tiêu biểu bao trùm các case khó — refund, null, duplicate key, ngày ở biên) vào repo, chạy toàn bộ pipeline trên nó trong môi trường test, rồi so sánh (diff) bảng output thực tế với fixture output kỳ vọng:

# test_pipeline_e2e.py
import pandas as pd
from pipeline import run_pipeline

def test_pipeline_produces_expected_daily_revenue(tmp_path):
    run_pipeline(
        input_path="fixtures/orders_sample.csv",
        output_path=tmp_path / "daily_revenue.csv",
    )

    actual = pd.read_csv(tmp_path / "daily_revenue.csv")
    expected = pd.read_csv("fixtures/expected_daily_revenue.csv")

    pd.testing.assert_frame_equal(
        actual.sort_values("order_date").reset_index(drop=True),
        expected.sort_values("order_date").reset_index(drop=True),
    )

Test này bắt được một loại bug mà unit test về bản chất bỏ sót: bug trong cách các bước ghép lại với nhau. Một join có thể đúng khi xét riêng lẻ nhưng vẫn tạo ra fan-out duplication một khi được nối vào toàn bộ DAG; chỉ một lần chạy end-to-end trên dữ liệu đã biết mới lộ ra điều đó.

Data-specific testing: phần thực sự mới

Đây là nhóm không có đối trọng thật sự trong application testing, vì nó không test logic chút nào — nó test chính dữ liệu, liên tục, khi dữ liệu chảy qua pipeline. Những test này không chỉ chạy lúc phát triển mà còn chạy trên mỗi lần chạy production, vì mục đích của chúng là bắt drift của thế giới thực, không phải regression của code.

Schema test khẳng định shape của dữ liệu không âm thầm thay đổi: các cột kỳ vọng vẫn tồn tại, không có cột lạ nào bỗng dưng biến mất, kiểu dữ liệu khớp (một cột string không âm thầm nhận int), và thứ tự cột hay ràng buộc nullability vẫn đúng như downstream consumer giả định.

Not-null, uniqueness và referential-integrity test là những test “cần cù” nhất của data testing — tương đương trực tiếp ở tầng dữ liệu của các database constraint, chỉ khác là được chạy tường minh dưới dạng test thay vì trông cậy vào constraint được enforce (các hệ thống nguồn và bảng staging thường không thể enforce chúng lúc ghi). dbt biến những test này thành declarative, gắn trực tiếp vào schema definition của model:

# models/staging/schema.yml
version: 2

models:
  - name: stg_orders
    columns:
      - name: order_id
        description: "Primary key of the orders staging table."
        tests:
          - not_null
          - unique
      - name: customer_id
        description: "Foreign key to the customers dimension."
        tests:
          - not_null
          - relationships:
              to: ref('stg_customers')
              field: customer_id
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'paid', 'refunded', 'cancelled']

dbt test chạy toàn bộ test trong project (hoặc một tập con với dbt test --select stg_orders) và fail build ngay khi một constraint bị vi phạm trên dữ liệu thật — ví dụ ngay khi xuất hiện một order row với customer_id không có row nào khớp trong stg_customers, điều mà unit test trên code transformation sẽ không bao giờ thấy vì logic biến đổi vẫn ổn; dữ liệu mới là thứ bị hỏng.

Distribution và anomaly test bắt một dạng lỗi khác: không có gì thực sự invalid, nhưng shape của dữ liệu lại đáng ngờ. Một feed nguồn thường ngày trả về 80.000-120.000 row bỗng chỉ trả về 8 row, hoặc 800.000 row, gần như luôn luôn là một bug — một lần extract chỉ lấy một phần, một filter bị lật sai, một join bị duplicate chạy loạn — dù mỗi row riêng lẻ vẫn pass schema check và null check.

-- a simple row-count anomaly test, expressible as a dbt singular test
-- tests/assert_order_volume_is_in_range.sql
select
    count(*) as row_count
from {{ ref('stg_orders') }}
where order_date = current_date - 1
having count(*) not between 50000 and 150000

Một dbt singular test như trên chỉ pass khi nó trả về zero row — một kết quả không rỗng bị coi là fail — đó là lý do query được viết sao cho chỉ trả về row đúng lúc điều kiện anomaly là đúng (having ... not between ...).

Business-rule test mã hóa kiến thức nghiệp vụ mà không constraint tổng quát nào diễn đạt được: revenue không bao giờ được âm ngoại trừ luồng refund tường minh, end_date của một subscription không bao giờ được đứng trước start_date, một discount không bao giờ được vượt quá 100% giá niêm yết. Đây thường là những test có giá trị cao nhất trên mỗi dòng code, vì chúng là những test bắt được đúng loại bug mà một stakeholder thực sự sẽ để ý.

models:
  - name: fct_orders
    columns:
      - name: net_revenue
        tests:
          - dbt_utils.expression_is_true:
              expression: ">= 0"

Idempotency testing

Một pipeline là idempotent nếu chạy cùng một job hai lần, trên cùng một input, cho ra cùng một end state như khi chỉ chạy một lần — không duplicate row, không double-count revenue, không làm hỏng aggregate. Mối quan tâm này gần như không tồn tại trong testing application stateless (một GET request chạy hai lần chỉ đơn giản trả về cùng dữ liệu hai lần), nhưng nó lại là trung tâm của testing pipeline vì pipeline thường xuyên retry: một scheduler trigger lại một run đã fail, một engineer chạy tay lại một backfill, một orchestrator retry một task sau một lỗi network thoáng qua. Nếu pipeline không idempotent, mỗi sự kiện vận hành bình thường đó sẽ âm thầm nhân đôi dữ liệu. Xem ETL/ELT & Data Pipelines để biết các design pattern (upsert trên natural key, INSERT OVERWRITE trên một partition, delete-rồi-insert trong một transaction) giúp idempotency khả thi — phần này nói về việc test rằng tính chất đó thực sự đúng.

Bản thân test này mô tả thì đơn giản đến bất ngờ nhưng lại rất dễ làm sai khi implement: chạy job, ghi lại state kết quả, chạy đúng job đó lần nữa trên đúng input đó, và assert state không đổi.

# test_idempotency.py
from pipeline import run_daily_load

def test_running_pipeline_twice_does_not_duplicate_rows(warehouse_conn, fixture_orders):
    run_daily_load(warehouse_conn, source_rows=fixture_orders, run_date="2026-01-10")
    first_count = warehouse_conn.execute(
        "SELECT count(*) FROM fct_orders WHERE load_date = '2026-01-10'"
    ).fetchone()[0]

    # re-run the identical job against the identical input, exactly as a retry would
    run_daily_load(warehouse_conn, source_rows=fixture_orders, run_date="2026-01-10")
    second_count = warehouse_conn.execute(
        "SELECT count(*) FROM fct_orders WHERE load_date = '2026-01-10'"
    ).fetchone()[0]

    assert first_count == second_count, "re-running the load duplicated rows"

Một phiên bản mạnh hơn của cùng test này còn hash hoặc checksum các row kết quả (chứ không chỉ đếm số lượng) để bắt một bug tinh vi hơn: số row không đổi, nhưng giá trị âm thầm thay đổi ở lần chạy thứ hai (ví dụ một aggregate bị cộng dồn thay vì bị ghi đè). Idempotency test nên nằm chung nhóm với integration test, vì chúng cần một đích lưu trữ thật, có state, mới thực sự có ý nghĩa — một fake in-memory reset giữa các lần gọi sẽ che giấu chính xác cái bug mà test này tồn tại để bắt.

Smoke testing và load testing cho pipeline

Một smoke test cho pipeline không nói về correctness theo nghĩa data quality — đó là cách kiểm tra nhanh nhất có thể rằng pipeline có chạy hay không sau một lần deploy: job có start không, có chạy đến cuối mà không có unhandled exception không, có ghi được ít nhất một row không, DAG có hoàn thành trong khung thời gian kỳ vọng không. Nó nên chạy trong vài giây và bắt được những hỏng hóc thảm khốc (sai credentials, thiếu dependency, lỗi syntax mà unit test bỏ sót vì module chưa từng được import) trước khi ai đó phải chờ một lần chạy data-quality đầy đủ hoàn tất.

Load testing một pipeline xác nhận nó sống sót ở peak volume thực tế, không chỉ ở volume vừa phải dùng trong integration test hàng ngày — tương đương với một đợt tăng traffic kiểu Black Friday cho một API e-commerce. Cụ thể, điều này nghĩa là đưa vào pipeline một khối lượng dữ liệu test cao gấp nhiều lần volume ngày thường và xác nhận nó hoàn thành trong SLA mà không hết bộ nhớ, không timeout ở connection pool database, và không sụp ở bước shuffle/join chỉ trở thành bottleneck khi ở quy mô lớn. Với batch pipeline, việc này thường được thực hiện bằng cách replay một dataset synthetic hoặc subsetted có volume cao qua một môi trường staging có kích thước tương đương production; với streaming pipeline, các công cụ như một script Kafka producer hoặc một load-generation harness đẩy message với throughput gấp nhiều lần bình thường để xác nhận consumer lag vẫn bị chặn trong ngưỡng và autoscaling (nếu có cấu hình) thực sự kích hoạt.

Testing ở quy mô production mà không dùng dữ liệu production

Testing ở volume thực tế gặp phải một ràng buộc cứng mà application testing hiếm khi gặp: bạn thường không thể đơn giản copy dữ liệu production vào môi trường test, vì lý do privacy, compliance (PII, GDPR/HIPAA), và cả vì khối lượng dữ liệu quá lớn, nhưng một bộ test chỉ từng thấy một fixture 20 row viết tay sẽ bỏ sót những bug chỉ xuất hiện ở quy mô lớn (một join fan-out ngoài dự kiến trên toàn bộ không gian key, một sort phải spill ra đĩa, một partition bị skew làm nghẽn một worker). Vài chiến lược bổ sung có thể thu hẹp khoảng cách đó:

Chiến lược	Cách hoạt động	Phù hợp nhất cho
Sampling	Lấy một mẫu ngẫu nhiên hoặc stratified có tính đại diện thống kê của dữ liệu production (ví dụ 1% số row, giữ nguyên phân phối của các key) vào môi trường test/staging.	Test volume và skew mà không phơi bày toàn bộ PII (vẫn cần masking).
Data subsetting	Dùng một công cụ nhận biết referential-integrity để lấy ra một lát cắt nhất quán của database quan hệ — ví dụ toàn bộ row cho một mẫu customer, trên tất cả bảng liên quan — thay vì lấy mẫu ngẫu nhiên độc lập trên từng bảng.	Integration/E2E test cần dữ liệu nhất quán, join được, trên nhiều bảng.
Synthetic data generation	Sinh dữ liệu bằng chương trình sao cho khớp schema production và các thuộc tính thống kê (phân phối, cardinality, quan hệ referential) mà không dùng bất kỳ record thật nào.	Các domain nhạy cảm về privacy, và stress-test các edge case mà dữ liệu thật có thể tình cờ không có.
Masking / anonymization	Copy đúng cấu trúc và volume của production, nhưng scrub hoặc tokenize các field nhạy cảm tại chỗ.	Khi tính thực tế của giá trị thật ít quan trọng hơn tính thực tế của volume và shape.

Một số công cụ đáng biết trong mảng này: các utility sample có sẵn của dbt và seed-based fixture cho các test nhỏ tất định, sampling native của database (TABLESAMPLE trong Postgres/Snowflake/BigQuery), các công cụ subsetting chuyên dụng như Tonic.ai hay data diffing của Datafold, và các thư viện synthetic-data như package Faker của Python hoặc SDV (Synthetic Data Vault) để tạo dataset giả lập nhưng thực tế về mặt thống kê. Dù dùng chiến lược nào, hãy coi môi trường test quy mô production đó là disposable và có thể rebuild bất cứ lúc nào — nó không bao giờ nên trở thành một bản sao “bóng” của production mà ai đó quên refresh hoặc bảo mật.

Tích hợp CI cho pipeline test

Không có nhóm test nào ở trên mang lại nhiều giá trị nếu con người phải tự nhớ để chạy chúng. Trong thực tế, dbt test và các bộ pytest bao phủ logic transformation và idempotency được nối vào CI để tự động chạy trên mỗi pull request đụng đến code pipeline hay định nghĩa model, chặn merge nếu bất kỳ test nào fail — cùng nguyên tắc “test như một merge gate” từ Backend — Testing, áp dụng cho dữ liệu. Cơ chế sâu hơn để xây dựng luồng CI/CD đặc thù cho pipeline đó (chạy test trên schema warehouse ephemeral theo từng branch, slim CI với dbt build --select state:modified+, và gate việc deploy dựa trên kết quả data test) được trình bày ở CI/CD for Data Engineering.

Loại test → kiểm tra điều gì → công cụ ví dụ

Loại test	Kiểm tra điều gì	Công cụ ví dụ
Unit	Logic của một transformation function riêng lẻ, độc lập	pytest, unittest
Integration	Dữ liệu di chuyển đúng giữa hệ thống nguồn và đích thật (hoặc containerized)	Testcontainers, pytest fixture trên sandbox schema
Functional / E2E	Toàn bộ pipeline cho ra output đúng về nghiệp vụ trên một dataset đã biết	pytest + fixture CSV, dbt seed
Schema	Cột tồn tại, kiểu dữ liệu và nullability khớp kỳ vọng	dbt schema test, Great Expectations, Pandera
Not-null / uniqueness / referential integrity	Ràng buộc ở cấp row đúng trên dữ liệu thật	dbt (not_null, unique, relationships)
Distribution / anomaly	Số lượng row và phân phối giá trị nằm trong khoảng kỳ vọng	dbt singular test, Great Expectations, Monte Carlo, Anomalo
Business-rule	Các bất biến đặc thù nghiệp vụ (ví dụ revenue ≥ 0) luôn đúng	dbt (dbt_utils.expression_is_true), custom SQL assertion
Idempotency	Chạy lại job không duplicate hay làm hỏng dữ liệu	pytest trên một staging warehouse thật
Smoke	Pipeline có chạy được sau deploy hay không	Health-check DAG nhẹ trên Airflow/Dagster, CI post-deploy job
Load	Pipeline sống sót ở peak volume trong SLA	Dataset synthetic volume cao, script load Kafka producer

Best Practices

• Test dữ liệu, không chỉ test code. Dành thời gian tường minh cho schema, null/uniqueness, distribution, và business-rule test — một bộ unit test xanh không nói lên được gì về việc dữ liệu hôm nay có hợp lý hay không.
• Không bao giờ test trực tiếp trên dữ liệu production. Dùng sandboxed schema, Testcontainers, sampling, subsetting, hoặc synthetic generation thay thế; truy cập production cho một lần chạy test là một rủi ro compliance và blast-radius, không phải một đường tắt.
• Biến idempotency thành một thuộc tính hạng nhất, được test tường minh, không phải một giả định. Viết một test chạy pipeline hai lần trên cùng input và assert output giống nhau, cho mọi job ghi vào shared state.
• Fail thật rõ ràng và cụ thể. Một distribution test chỉ nói “anomaly detected” có giá trị thấp hơn nhiều so với một test báo “expected 50k-150k rows, got 8” — hãy làm cho thông báo lỗi có thể hành động được.
• Giữ data test chạy liên tục trên production, không chỉ trong CI. Schema và distribution drift xảy ra ở thế giới thực lúc 3 giờ sáng, không chỉ khi có PR đang mở; các lần chạy dbt test theo lịch trên dữ liệu production (không chỉ pre-merge) bắt được drift từ upstream mà không PR nào bắt được.
• Ưu tiên subsetting hoặc synthetic data hơn sampling tràn lan khi các bảng có quan hệ foreign-key — 1% mẫu độc lập của từng bảng gần như không bao giờ join sạch với nhau.
• Thêm một smoke test nhanh làm stage đầu tiên của CI/deploy, trước khi bộ data-quality đầy đủ chạy, để hỏng hóc thảm khốc fail trong vài giây, chứ không phải sau mười phút chạy test.
• Viết một regression fixture cho mỗi bug dữ liệu thật mà bạn sửa, giống như bạn làm với bug application — đúng shape row đã gây ra sự cố trở thành một fixture cố định trong dataset E2E test.
• Gate việc merge dựa trên test, không dựa trên “chạy ổn ở dev.” Nối dbt test và pytest vào CI như các required check, theo CI/CD for Data Engineering, để một pipeline hỏng không thể lên production chỉ vì được đóng dấu xanh qua loa.

Tài liệu tham khảo

• dbt — About dbt tests
• Great Expectations — Documentation
• Pandera — Statistical data testing for pandas
• Martin Fowler — Test Pyramid
• Testcontainers — Documentation
• SDV — Synthetic Data Vault
• Monte Carlo — Data Observability and Anomaly Detection
• pytest — Documentation

# transforms.py
from datetime import date

def normalize_order(raw: dict) -> dict:
    """Coerce a raw order record into the canonical shape used downstream."""
    if raw.get("amount_cents") is None:
        raise ValueError("amount_cents is required")

    return {
        "order_id": str(raw["order_id"]),
        "amount": round(raw["amount_cents"] / 100, 2),
        "currency": (raw.get("currency") or "USD").upper(),
        "order_date": date.fromisoformat(raw["order_date"]),
        "is_refund": raw.get("amount_cents", 0) < 0,
    }

# test_transforms.py
import pytest
from datetime import date
from transforms import normalize_order

def test_converts_cents_to_decimal_amount():
    raw = {"order_id": 1, "amount_cents": 1999, "currency": "usd", "order_date": "2026-01-05"}
    result = normalize_order(raw)
    assert result["amount"] == 19.99

def test_defaults_missing_currency_to_usd():
    raw = {"order_id": 2, "amount_cents": 500, "order_date": "2026-01-05"}
    result = normalize_order(raw)
    assert result["currency"] == "USD"

def test_flags_negative_amounts_as_refunds():
    raw = {"order_id": 3, "amount_cents": -1000, "order_date": "2026-01-05"}
    result = normalize_order(raw)
    assert result["is_refund"] is True

def test_raises_when_amount_is_missing():
    raw = {"order_id": 4, "order_date": "2026-01-05"}
    with pytest.raises(ValueError):
        normalize_order(raw)

# test_pipeline_integration.py
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.kafka import KafkaContainer
from pipeline import extract_new_orders, load_orders

@pytest.fixture(scope="module")
def source_db():
    with PostgresContainer("postgres:16-alpine") as pg:
        conn = pg.get_connection_url()
        # seed a source schema that mimics production shape
        yield conn

@pytest.fixture(scope="module")
def warehouse_db():
    with PostgresContainer("postgres:16-alpine") as pg:
        yield pg.get_connection_url()

def test_pipeline_moves_new_orders_end_to_end(source_db, warehouse_db):
    rows = extract_new_orders(source_db, since="2026-01-01")
    load_orders(warehouse_db, rows)

    # assert against the real destination, not a mock
    import psycopg
    with psycopg.connect(warehouse_db) as conn:
        count = conn.execute("SELECT count(*) FROM orders").fetchone()[0]
        assert count == len(rows)

# test_pipeline_e2e.py
import pandas as pd
from pipeline import run_pipeline

def test_pipeline_produces_expected_daily_revenue(tmp_path):
    run_pipeline(
        input_path="fixtures/orders_sample.csv",
        output_path=tmp_path / "daily_revenue.csv",
    )

    actual = pd.read_csv(tmp_path / "daily_revenue.csv")
    expected = pd.read_csv("fixtures/expected_daily_revenue.csv")

    pd.testing.assert_frame_equal(
        actual.sort_values("order_date").reset_index(drop=True),
        expected.sort_values("order_date").reset_index(drop=True),
    )

# models/staging/schema.yml
version: 2

models:
  - name: stg_orders
    columns:
      - name: order_id
        description: "Primary key of the orders staging table."
        tests:
          - not_null
          - unique
      - name: customer_id
        description: "Foreign key to the customers dimension."
        tests:
          - not_null
          - relationships:
              to: ref('stg_customers')
              field: customer_id
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'paid', 'refunded', 'cancelled']

-- a simple row-count anomaly test, expressible as a dbt singular test
-- tests/assert_order_volume_is_in_range.sql
select
    count(*) as row_count
from {{ ref('stg_orders') }}
where order_date = current_date - 1
having count(*) not between 50000 and 150000

models:
  - name: fct_orders
    columns:
      - name: net_revenue
        tests:
          - dbt_utils.expression_is_true:
              expression: ">= 0"

# test_idempotency.py
from pipeline import run_daily_load

def test_running_pipeline_twice_does_not_duplicate_rows(warehouse_conn, fixture_orders):
    run_daily_load(warehouse_conn, source_rows=fixture_orders, run_date="2026-01-10")
    first_count = warehouse_conn.execute(
        "SELECT count(*) FROM fct_orders WHERE load_date = '2026-01-10'"
    ).fetchone()[0]

    # re-run the identical job against the identical input, exactly as a retry would
    run_daily_load(warehouse_conn, source_rows=fixture_orders, run_date="2026-01-10")
    second_count = warehouse_conn.execute(
        "SELECT count(*) FROM fct_orders WHERE load_date = '2026-01-10'"
    ).fetchone()[0]

    assert first_count == second_count, "re-running the load duplicated rows"