DataOps nâng cao 12 — Nối DataOps với MLOps

Vì sao MLOps phải đứng trên nền DataOps

Có một sự thật mà đội khoa học dữ liệu hay bỏ qua: mô hình chỉ tốt bằng dữ liệu nuôi nó. Bạn có thể tinh chỉnh hyperparameter cả tuần, thay XGBoost bằng mạng nơ-ron, nhưng nếu feature bị rò rỉ nhãn tương lai (label leakage), bị lệch phân phối giữa train và serving, hoặc không tái tạo được — thì mọi con số AUC đẹp đẽ đều là ảo. Đó là lý do DataOps là NỀN của MLOps: pipeline dữ liệu tin cậy, có test, có observability và có lineage chính là điều kiện cần để MLOps chạy được trong sản xuất, đặc biệt ở ngân hàng nơi mỗi mô hình tín dụng/gian lận phải qua model risk validation.

Bài này không dạy lại MLOps từ đầu (xem ML — Workflow, MLOps — Feature Store, MLOps — Pipelines). Nó tập trung vào đường ranh giới giữa hai thế giới và cách một đội DataOps ngân hàng phục vụ đội ML: xây feature như một sản phẩm dữ liệu, đảm bảo point-in-time correctness, reproducibility cho audit, CI/CD và orchestration chung, và lineage giải trình được.

Ranh giới & phối hợp DataOps ⟷ MLOps

Silo giữa hai đội là nguồn gốc của phần lớn sự cố mô hình. Cách chia trách nhiệm thực dụng:

Điểm giao thoa quan trọng nhất là feature và contract feature — nơi hai đội phải cùng ký một hợp đồng: đội ML nói "tôi cần txn_count_30d của account tính đến mốc dự đoán, kiểu integer, không null, tính as-of"; đội DataOps cam kết cung cấp đúng ngữ nghĩa đó, có test giữ đúng và có SLA về freshness. Đây chính là data contract áp cho ML.

Feature pipeline như một sản phẩm DataOps

Đây là ý tưởng cốt lõi: feature không phải cột phụ, nó là sản phẩm dữ liệu. Nghĩa là feature phải được xây bằng cùng bộ công cụ và kỷ luật như mọi bảng mart khác — dbt/pipeline có test, có observability — rồi mới đẩy vào feature store.

Xây feature bằng dbt/pipeline có test

Một feature model dbt minh hoạ cho mô hình gian lận (chỉ là ví dụ, không phải cột thật của sandbox):

-- models/features/feat_account_txn_30d.sql  (MINH HOẠ, dbt)
-- Tính as-of theo tham số ref_date để tránh rò rỉ tương lai
with base as (
    select
        a.id                                as account_id,
        t.amount,
        t.created_at
    from {{ ref('stg_accounts') }} a
    join {{ ref('stg_transactions') }} t on t.account_id = a.id
    where t.created_at <  '{{ var("ref_date") }}'::timestamp          -- as-of: chỉ dữ liệu TRƯỚC mốc
      and t.created_at >= '{{ var("ref_date") }}'::timestamp - interval '30 days'
)
select
    account_id,
    '{{ var("ref_date") }}'::timestamp        as as_of_ts,
    count(*)                                    as txn_count_30d,
    coalesce(sum(amount), 0)                    as txn_sum_30d,
    coalesce(avg(amount), 0)                    as txn_avg_30d
from base
group by account_id

Kèm theo là các test (schema test + singular test kiểu Quality & Contracts): txn_count_30d >= 0, account_id không null và unique theo (account_id, as_of_ts), và một test đối soát rằng không có giao dịch nào created_at >= as_of_ts lọt vào phép tính. Test cuối chính là hàng rào chống label leakage.

Point-in-time correctness & tránh rò rỉ nhãn tương lai

Sai lầm kinh điển: tính feature "tổng giao dịch 30 ngày" bằng toàn bộ dữ liệu hiện có, rồi ghép với nhãn "account này gian lận vào ngày X". Khi làm vậy, feature vô tình chứa cả giao dịch sau ngày X — tức là thông tin từ tương lai mà lúc chấm điểm thật ta không thể có. Mô hình học được "bí kíp" này sẽ đẹp lúc backtest và sập lúc lên production.

Cách đúng là as-of / point-in-time: mọi feature phải được tính chỉ từ dữ liệu tồn tại tại mốc dự đoán. Với train, mỗi mẫu có một ref_date riêng (thời điểm ra quyết định); feature phải cắt đúng created_at < ref_date. Đây là lý do feature store offline lưu feature theo thời điểm và hỗ trợ point-in-time join, còn feature store online chỉ giữ giá trị mới nhất để serving độ trễ thấp (xem Feature Store).

Dưới đây là feature point-in-time chạy được ngay trên sandbox: tổng và đếm giao dịch 30 ngày gần nhất tính đến một mốc (as-of) cho từng account. Chú ý điều kiện created_at < mốc — chính là ranh giới chống rò rỉ tương lai:

-- ▶ Chạy được
WITH as_of AS (SELECT TIMESTAMP '2024-06-01 00:00:00' AS ref_ts)
SELECT
  a.id                                                   AS account_id,
  (SELECT ref_ts FROM as_of)                             AS as_of_ts,
  COUNT(t.id)                                            AS txn_count_30d,
  COALESCE(SUM(t.amount), 0)                             AS txn_sum_30d,
  ROUND(COALESCE(AVG(t.amount), 0)::numeric, 2)          AS txn_avg_30d
FROM accounts a
LEFT JOIN transactions t
       ON t.account_id = a.id
      AND t.created_at <  (SELECT ref_ts FROM as_of)                          -- as-of: KHÔNG lấy tương lai
      AND t.created_at >= (SELECT ref_ts FROM as_of) - INTERVAL '30 days'
GROUP BY a.id
ORDER BY txn_sum_30d DESC
LIMIT 20;

Đổi ref_ts bạn sẽ có "ảnh chụp" feature tại từng mốc — đúng tinh thần point-in-time. Đây là khác biệt sống còn với một transform báo cáo BI thông thường (vốn chỉ quan tâm "hiện tại").

Đẩy vào feature store offline/online

Sau khi tính, feature được ghi hai chiều: offline (kho lớn, theo lịch sử, phục vụ train + batch scoring) và online (key-value độ trễ thấp, phục vụ serving realtime). DataOps chịu trách nhiệm materialize — batch job ghi feature mới nhất vào online store, và snapshot theo thời điểm vào offline store. Sự nhất quán offline↔online (cùng một công thức tính) là điều kiện tránh train/serving skew.

Dữ liệu huấn luyện reproducible

Ở ngân hàng, câu hỏi kiểm toán không phải "mô hình có tốt không" mà là "chứng minh tập train của phiên bản mô hình đang chạy được sinh ra như thế nào". Muốn trả lời được, tập train phải tái tạo được (reproducible). Tái tạo cần version hoá đủ ba thứ:

• Dữ liệu: snapshot / data version (VD Delta time-travel — xem Delta Lake, hoặc bảng snapshot bất biến) ứng với ngày build.
• Code: git commit của feature pipeline + code train.
• Config: ref_date, danh sách feature, tham số cắt mẫu, seed.

Gộp lại thành một training manifest (một bản ghi khai báo mọi input), lưu cùng model trong registry. Khi validator hỏi, ta chạy lại đúng manifest và ra đúng tập train — không "tôi nhớ là...". Đây là điều CI/CD cho Analytics và MLOps Pipelines cùng nhấn mạnh: reproducibility = data version + code version + config version.

Training / Inference data pipelines

Cần tách bạch ba loại pipeline, đừng trộn với transform báo cáo:

1. Training data pipeline: point-in-time join feature ↔ nhãn, sinh tập train reproducible. Chạy theo đợt (khi retrain).
2. Batch scoring pipeline: mô hình đã deploy chấm điểm hàng loạt (VD chấm rủi ro toàn bộ account mỗi đêm) và ghi kết quả vào bảng để hệ thống nghiệp vụ đọc. Đây là một ELT bình thường có thêm bước gọi model.
3. Serving data pipeline: cấp feature online mới nhất cho endpoint realtime (VD chấm gian lận ngay khi giao dịch phát sinh).

Điểm dễ sai: nhét logic feature vào cùng dbt model dùng cho báo cáo quản trị. Feature cho ML có ràng buộc as-of và ngữ nghĩa riêng; báo cáo BI thì tổng hợp "đến hiện tại". Trộn chung khiến một bên sửa làm bên kia rò rỉ. Tách rõ layer feature khỏi layer report.

Chất lượng & observability cho dữ liệu ML

Đội DataOps giám sát data drift ở đầu vào feature — phân phối feature hôm nay có lệch so với lúc train không. Đây là nối dài của Observability & DRE: freshness, volume, null rate, và thêm distribution drift của các feature quan trọng.

Phân biệt rạch ròi hai khái niệm hay bị nhầm:

Ví dụ giám sát data drift: so số dư trung bình theo account ở kỳ gần nhất với kỳ trước. Feature "số dư trung bình" là đầu vào phổ biến cho mô hình tín dụng; nếu phân phối của nó nhảy bậc, cần cảnh báo trước khi mô hình chấm sai. Câu sau chạy được trên sandbox, tính số dư trung bình as-of theo tháng để soi dịch chuyển phân phối:

-- ▶ Chạy được
WITH bucket AS (
  SELECT
    a.id                                          AS account_id,
    a.currency,
    date_trunc('month', t.created_at)             AS mth,
    AVG(a.balance)                                AS avg_balance,   -- feature "số dư TB" as-of theo tháng
    COUNT(t.id)                                   AS txns
  FROM accounts a
  JOIN transactions t ON t.account_id = a.id
  GROUP BY a.id, a.currency, date_trunc('month', t.created_at)
)
SELECT
  mth,
  currency,
  COUNT(*)                                        AS n_accounts,
  ROUND(AVG(avg_balance)::numeric, 2)             AS mean_feature,
  ROUND(STDDEV_POP(avg_balance)::numeric, 2)      AS std_feature
FROM bucket
GROUP BY mth, currency
ORDER BY mth, currency;

So mean_feature/std_feature giữa các tháng cho thấy phân phối feature có trôi không — nền tảng để dựng alert drift đúng nghĩa DRE.

CI/CD chung: gate chất lượng dữ liệu trước train

Pipeline dữ liệu và pipeline ML dùng chung một triết lý CI/CD: mọi thay đổi (feature logic, model code) đều qua PR, test tự động, và cổng chất lượng. Khác biệt: trong ML, có một gate chất lượng dữ liệu bắt buộc chạy TRƯỚC bước train. Nếu feature không đạt test (null vượt ngưỡng, drift lớn, freshness trễ) thì chặn retrain — vì train trên dữ liệu bẩn còn tệ hơn không train.

Chi tiết CI/CD phần ML ở MLOps Pipelines; phần data ở CI/CD Analytics. Điểm nối là gate DQ giữa hai bên.

Orchestration & lineage chung

Một orchestrator duy nhất (thường là Airflow) chạy cả chuỗi: ELT → feature → train/score. DAG chung giúp Airflow biết feature phụ thuộc ELT, train phụ thuộc feature — nên khi ELT trễ, train tự chờ, không train trên dữ liệu thiếu.

ELT (dbt run) ──▶ feature materialize ──▶ [train]  (khi retrain)
                                     └──▶ [batch score] ──▶ ghi bảng kết quả

Lineage khép vòng giải trình: nguồn → feature → mô hình → quyết định. Khi một khoản vay bị từ chối bị khiếu nại, hoặc validator hỏi "feature này lấy từ đâu", ta lần ngược được: quyết định ← model version X ← tập train (manifest) ← feature txn_count_30d ← bảng transactions chi nhánh Y. Đây là điều kiện tuân thủ, không phải tính năng "cho vui" (xem Catalog & Lineage).

Use case thực tế

Bối cảnh: NCB xây mô hình phát hiện gian lận thẻ. Đội ML yêu cầu 3 feature cho mỗi giao dịch tại thời điểm phát sinh: txn_count_30d, txn_sum_30d, avg_balance của account — tất cả as-of tại mốc giao dịch.

Vấn đề gặp phải (thật): phiên bản v1 tính feature bằng toàn bộ lịch sử giao dịch. Backtest cho AUC 0.94 — quá đẹp. Khi validator model risk soi, họ phát hiện txn_count_30d của một giao dịch ngày 1/6 đang đếm cả giao dịch ngày 15/6 (rò rỉ tương lai). Đó là label leakage cổ điển.

Cách DataOps xử lý:

1. Viết lại feature model theo point-in-time, cắt created_at < ref_ts (đúng logic 2 block SQL ở trên).
2. Thêm singular test đối soát: đếm số dòng feature có bất kỳ giao dịch created_at >= as_of_ts — kỳ vọng 0. Test này thành gate DQ trong CI, chặn train nếu > 0.
3. Đóng gói training manifest: data snapshot ngày build + git commit feature pipeline + ref_date mỗi mẫu → lưu cùng model trong registry để tái tạo cho audit.
4. Bật drift monitor cho avg_balance và txn_sum_30d theo tháng (block SQL số 2), alert khi lệch > 30% baseline.

Kết quả: sau khi bỏ leakage, AUC thực tế về 0.87 — thấp hơn nhưng thật, và mô hình không sập khi lên production. Quan trọng hơn, khi NHNN kiểm tra, đội chạy lại manifest ra đúng tập train, chỉ được lineage nguồn→feature→mô hình→quyết định. Toàn bộ gói validation gọn trong nửa ngày thay vì "đi tìm lại dữ liệu" hàng tuần.

Ghi nhớ

• DataOps là nền của MLOps: mô hình chỉ tốt khi dữ liệu tin cậy — ưu tiên chất lượng feature trước khi tuning mô hình.
• Feature là sản phẩm dữ liệu: xây bằng dbt/pipeline có test + observability, rồi đẩy vào feature store offline (train) và online (serving).
• Point-in-time / as-of là bắt buộc: luôn cắt created_at < mốc dự đoán để tránh rò rỉ nhãn tương lai (label leakage) — nguyên nhân #1 của mô hình "đẹp mà sập".
• Reproducible = data version + code version + config version, gói thành training manifest lưu cùng model — điều kiện sống còn cho audit ngân hàng.
• Tách feature khỏi report: đừng nhét logic feature vào transform BI; ba pipeline train / batch score / serving có ràng buộc khác nhau.
• Data drift (DataOps) ≠ concept drift (MLOps): một cái theo dõi phân phối đầu vào feature, một cái theo dõi quan hệ X→y của mô hình.
• CI/CD chung có gate DQ trước train; orchestration chung (Airflow) chạy ELT→feature→train/score; lineage khép vòng nguồn→feature→mô hình→quyết định để giải trình.
• Chống silo bằng contract feature: đội ML khai báo cần gì, đội DataOps cam kết ngữ nghĩa + SLA.