Thực hành: dựng năm lớp kiểm thử để chặn dữ liệu bẩn của khách trước khi tới model

Thử hình dung tuần đầu tiên ở một chuỗi bán lẻ. Khách gửi file đơn hàng xuất từ ba hệ thống. Ở cột amount, có dòng ghi “1.250.000”, có dòng ghi “1,250,000”, có dòng để trống, lại có một mã đơn xuất hiện hai lần. Model dự báo doanh thu sẽ nhận hết những dòng đó mà không phàn nàn gì.

IBM gọi hiện tượng này bằng câu quen thuộc “garbage in, garbage out”: dữ liệu vào tồi thì model ra cũng sai. IBM cũng định nghĩa chất lượng dữ liệu là mức độ một tập dữ liệu đạt các tiêu chí chính xác, đầy đủ, hợp lệ và nhất quán. Bài này biến bốn tiêu chí đó thành code chạy được.

Với một FDE, phần việc này thường quyết định demo đầu tiên có giữ được niềm tin của khách hay không. Khi model trả số lạ, bạn cần chỉ ra được dòng dữ liệu nào gây ra chuyện đó và vì sao dòng ấy đã lọt qua.

Bạn sẽ dựng gì, và cần chuẩn bị gì?

Bạn sẽ dựng năm lớp kiểm tra theo tinh thần kim tự tháp kiểm thử. Lớp dưới cùng là nhiều unit test nhanh và rẻ. Lớp trên cùng là rất ít bài end-to-end (E2E), vì loại này chậm, đắt và dễ vỡ. BrowserStack nhấn mạnh rằng kim tự tháp chỉ gợi ý nên đặt test ở đâu, không bắt buộc một tỉ lệ cố định.

Bạn chỉ cần Python 3 và một thư mục trống. Code dưới đây cố ý dùng Python thuần, gồm assert và module csv, để bạn thấy rõ logic bên dưới. Đây là bản đơn giản hoá: pipeline chỉ nhận đơn bằng VND.

Năm bước đầu tương ứng với năm lớp. Bước 6 không phải lớp thứ sáu: đó là bước chuyển công cụ, đưa phần kiểm tra dữ liệu và cổng chặn sang một thư viện chuyên dụng khi dự án lớn lên.

Bước 1: unit test cho hàm biến đổi thuần

Bắt đầu từ hàm chuẩn hoá số tiền, nơi lỗi dễ chui vào nhất. Hàm dưới đây chỉ dành cho VND: nó bỏ mọi dấu chấm và dấu phẩy, vì tiền đồng không có phần thập phân.

# transform.py
def normalize_amount(raw):
    """Chỉ dùng cho VND. Không áp dụng cho tiền có phần thập phân."""
    if raw is None:
        return None
    cleaned = raw.strip().replace(".", "").replace(",", "")
    if not cleaned.isdigit():
        return None
    return float(cleaned)

# test_transform.py
from transform import normalize_amount

assert normalize_amount("1.250.000") == 1250000.0
assert normalize_amount("1,250,000") == 1250000.0
assert normalize_amount(" 500000 ") == 500000.0
assert normalize_amount("") is None
assert normalize_amount("abc") is None
print("unit OK")

Chạy python test_transform.py. Nếu thấy dòng unit OK là đạt. Guru99 giải thích giá trị của lớp này: unit test làm lỗi lộ ra gần chỗ nó sinh ra, nên sửa nhanh hơn và rẻ hơn.

Giới hạn VND là lựa chọn có chủ đích. Nếu đưa một đơn USD ghi “99.50” vào hàm này, kết quả sẽ là 9950, sai gấp 100 lần mà không báo lỗi gì. Vì thế ở bước 2, mọi dòng không phải VND sẽ bị gắn cờ và cách ly, cho tới khi bạn viết riêng một hàm chuẩn hoá cho từng loại tiền.

Bạn cũng nên thử normalize_amount("-50000"). Kết quả là None, vì dấu trừ làm isdigit() trả về sai. Đơn hoàn tiền có được phép âm hay không là một quyết định nghiệp vụ, và bạn phải hỏi khách chứ không tự đoán.

Bước 2: biến bốn tiêu chí chất lượng thành kiểm tra

Unit test kiểm tra code của bạn. Bước này kiểm tra dữ liệu của khách. Mỗi điều kiện bên dưới tương ứng với một tiêu chí của IBM.

# checks.py
SUPPORTED_CURRENCY = "VND"  # bản đơn giản hoá: chỉ nhận VND

def check_batch(rows):
    failures, seen = [], set()
    for i, r in enumerate(rows):
        oid = r.get("order_id")
        if not oid:
            failures.append((i, "thiếu order_id"))          # đầy đủ
        elif oid in seen:
            failures.append((i, "order_id trùng"))          # nhất quán
        seen.add(oid)
        if r.get("amount") is None or r["amount"] <= 0:
            failures.append((i, "amount không hợp lệ"))     # hợp lệ
        if r.get("currency") != SUPPORTED_CURRENCY:
            failures.append((i, "currency chưa hỗ trợ"))    # hợp lệ
    return failures

Tiêu chí chính xác khó viết thành luật nhất, vì một con số có thể đúng định dạng mà vẫn sai so với thực tế. Cách thực tế là xin khách một con số đối chiếu, chẳng hạn tổng doanh thu tháng trước trên hệ thống kế toán. Sau đó bạn viết thêm một kiểm tra so tổng của batch với con số ấy.

Bước 3: integration test giữa các stage

Từng hàm chạy đúng chưa có nghĩa là ghép lại sẽ đúng. Guru99 mô tả integration test là lớp dùng để phát hiện lỗi ở chỗ các module giao tiếp với nhau. Trong pipeline, đó là khớp nối giữa bước đọc, bước biến đổi và bước kiểm tra.

Hãy tạo fixtures/orders_sample.csv gồm năm dòng và cố tình cài hai lỗi:

order_id,amount,currency
A1,1.250.000,VND
A2,"1,250,000",VND
A2,300000,VND
A4,,VND
A5,99000,VND

# test_pipeline.py
import csv
from transform import normalize_amount
from checks import check_batch

def load(path):
    with open(path, newline="", encoding="utf-8") as f:
        rows = list(csv.DictReader(f))
    for r in rows:
        r["amount"] = normalize_amount(r.get("amount"))
    return rows

rows = load("fixtures/orders_sample.csv")
msgs = {m for _, m in check_batch(rows)}
assert len(rows) == 5
assert msgs == {"order_id trùng", "amount không hợp lệ"}
print("integration OK")

Test này bắt được loại lỗi mà unit test không thấy. Trong file, số tiền có dấu phẩy phải nằm trong ngoặc kép. Nếu một lần xuất file sau đó thiếu ngoặc kép, module csv sẽ tách “1,250,000” thành nhiều cột, cột currency nhận một giá trị lạ, và test báo đỏ ngay thay vì để con số sai đi tiếp.

Bước 4: cổng chặn cho mỗi lần chạy

Theo Guru99, smoke test xác nhận bản build đủ ổn để kiểm thử tiếp. Ở pipeline, bạn áp ý tưởng đó cho từng batch: batch đủ sạch thì cho qua, còn không thì dừng trước khi tới model.

# gate.py
from checks import check_batch

MAX_BAD_RATIO = 0.02  # ngưỡng giả định, cần thống nhất với khách

def gate(rows):
    bad = {i for i, _ in check_batch(rows)}
    ratio = len(bad) / max(len(rows), 1)
    if not rows or ratio > MAX_BAD_RATIO:
        raise SystemExit(f"CHẶN batch: {len(bad)}/{len(rows)} dòng lỗi")
    clean = [r for i, r in enumerate(rows) if i not in bad]
    quarantine = [r for i, r in enumerate(rows) if i in bad]
    return clean, quarantine

Giả sử một batch có 1.000 dòng. Nếu 15 dòng lỗi, tỉ lệ là 1,5%: batch đi tiếp với 985 dòng sạch, còn 15 dòng vào khu cách ly để khách xem lại. Nếu 37 dòng lỗi, tỉ lệ là 3,7%: cả batch bị chặn, vì lỗi nhiều cỡ đó thường cho thấy hệ thống nguồn đã thay đổi chứ không chỉ vài dòng gõ nhầm.

Bước 5: chỉ một vài bài E2E, chạy gần production

Microsoft Code-With Engineering Playbook mô tả E2E là kiểm tra toàn bộ ứng dụng từ đầu đến cuối với mọi thành phần ghép lại. Playbook cũng cảnh báo rằng test càng lên cao càng chậm, càng tốn công viết, chạy và bảo trì.

Vì vậy, một hoặc hai kịch bản là đủ: lấy file thật đã che dữ liệu nhạy cảm, chạy qua toàn bộ pipeline, rồi xem đầu ra của model có hợp lý không.

TestGrid khuyên luôn đặt góc nhìn của người dùng lên trước. Ở chỗ khách, lời khuyên hợp lý là chạy E2E càng gần production càng tốt: đúng kiểu kết nối, đúng quyền truy cập và đúng định dạng file mà hệ thống thật sẽ gửi. File mẫu bạn tự soạn trên laptop không thay được những thứ đó.

Bước 6: chuyển sang Great Expectations

Bước này không thêm lớp mới mà thay công cụ cho lớp kiểm tra dữ liệu (bước 2) và cổng chặn (bước 4). Khi số luật tăng lên, bạn nên chuyển phần dữ liệu sang Great Expectations.

GX Core là thư viện Python để xây dựng và chạy các luồng kiểm định dữ liệu bằng code. Mỗi điều kiện trong check_batch tương ứng với một Expectation, tức một khẳng định có thể kiểm chứng về dữ liệu.

Đoạn dưới đây là bản minh hoạ cho đúng một luật, “không được thiếu order_id”, viết theo cách gọi của GX Core 1.x. Tên hàm có thể khác giữa các phiên bản, nên hãy đối chiếu với tài liệu GX của bản bạn cài trước khi dùng.

# gx_demo.py  (minh hoạ, kiểm tra lại theo tài liệu GX)
import great_expectations as gx
import pandas as pd

df = pd.read_csv("fixtures/orders_sample.csv", dtype=str)

context = gx.get_context()
source = context.data_sources.add_pandas("orders")
asset = source.add_dataframe_asset(name="orders_df")
batch_def = asset.add_batch_definition_whole_dataframe("whole")
batch = batch_def.get_batch(batch_parameters={"dataframe": df})

exp = gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
print(batch.validate(exp).success)

Với fixture ở bước 3, kết quả nên là True, vì cả năm dòng đều có order_id. Hãy thử xoá mã ở một dòng rồi chạy lại để thấy kết quả chuyển sang False.

Hàm gate thì tương ứng với Checkpoint. Tài liệu của GX gọi Checkpoint là cách chính để kiểm định dữ liệu khi triển khai GX trong production. Bài tập tiếp theo cho bạn: tự viết ba luật còn lại ở bước 2 thành Expectation, rồi gom chúng vào một Checkpoint.

Những lỗi hay gặp

Lỗi phổ biến nhất là dồn sức vào E2E và bỏ qua unit test. Khi đó mỗi lần đỏ bạn mất cả buổi để tìm xem lỗi nằm ở bước nào. Một lỗi khác là tự đặt ngưỡng chặn mà không hỏi khách. Ngưỡng 2% là chuyện nghiệp vụ, không phải chuyện kỹ thuật.

Cuối cùng là âm thầm xoá dòng lỗi. Cách ly nghĩa là giữ lại dòng lỗi, ghi rõ lý do và gửi danh sách cho khách. Rất có thể khách sẽ phát hiện chính hệ thống của họ đang sinh dữ liệu sai.

Ghi kỹ năng này vào CV thế nào?

Trên CV, đừng chỉ ghi “viết test”. Hãy ghi một câu cụ thể, ví dụ: dựng cổng kiểm định cho batch, cách ly các dòng lỗi và chặn những batch vượt ngưỡng trước khi tới model.

Khi đi phỏng vấn, hãy mang theo đúng file fixture năm dòng ở bước 3 và sẵn sàng chạy thử ngay trên laptop. Cho người phỏng vấn thấy bạn cài lỗi nào vào dữ liệu, và pipeline bắt từng lỗi ở lớp nào.