DataOps nâng cao 20 — Dagster & Software-Defined Assets

Suốt series này, khi nói "điều phối" ta ngầm hiểu theo cách của Airflow: định nghĩa các task, nối chúng thành DAG, orchestrator lo thứ tự chạy. Đó là task-based orchestration — mô hình đã thống trị một thập kỷ. Nhưng có một paradigm khác đang lên, đảo ngược trọng tâm: thay vì hỏi "chạy task gì, theo thứ tự nào", ta hỏi "cần tồn tại những dữ liệu gì, và mỗi dữ liệu được tạo ra từ đâu". Đây là asset-based orchestration, và Dagster là công cụ đại diện. Bài này giải thích vì sao góc nhìn này quan trọng với một team dữ liệu ngân hàng cần lineage và freshness rõ ràng để giải trình.

Vấn đề của tư duy hướng tác vụ

Trong Airflow truyền thống, một DAG là tập các task và cạnh phụ thuộc. Task run_dbt chạy trước task refresh_dashboard. Orchestrator biết chính xác chạy gì và khi nào, nhưng nó không biết task tạo ra dữ liệu gì. Với nó, run_dbt chỉ là "một hộp đen sẽ chạy xong lúc nào đó"; bên trong hộp đó có bảng stg_giao_dich, mart_bao_cao_cuoi_ngay hay 40 bảng khác — orchestrator hoàn toàn mù.

Hệ quả thực tế đau đớn:

• Lineage rời rạc: DAG nói về thứ tự task, không phải dòng chảy dữ liệu. Muốn biết "bảng gold mart_du_no phụ thuộc bảng thô nào", bạn phải đọc code bên trong task, hoặc dựng công cụ lineage riêng. Đồ thị task ≠ đồ thị dữ liệu.
• Quan sát theo task, không theo dữ liệu: bạn biết "task chạy 3 phút, thành công", nhưng câu hỏi nghiệp vụ là "bảng mart_du_no được cập nhật lần cuối lúc nào, có tươi không". Task xanh không đảm bảo dữ liệu tươi.
• Chạy lại thô: muốn rebuild một bảng, bạn chạy lại cả task (có thể là cả dbt run toàn bộ), vì đơn vị điều phối là task chứ không phải bảng.
• Phụ thuộc ngầm: khi mart_A thật ra đọc từ stg_B nhưng hai bảng nằm ở hai DAG khác nhau, phụ thuộc dữ liệu đó vô hình với orchestrator. Đây là nguồn của vô số sự cố "dashboard hiện số cũ".

Không phải Airflow sai — nó bổ sung Datasets/Assets đúng để vá điểm này. Nhưng ở Airflow, asset là lớp phủ trên mô hình task. Ở Dagster, asset là primitive nền tảng.

Software-Defined Assets: khai báo dữ liệu, không phải công việc

Ý tưởng cốt lõi của Software-Defined Assets (SDA): mỗi tài sản dữ liệu — một bảng, một dataset, một file Parquet, một ML model — được khai báo trực tiếp trong code như một asset, gồm hai phần:

1. Định danh của tài sản (nó là bảng nào).
2. Hàm tính toán tạo ra nó, và những asset khác mà nó phụ thuộc.

Từ tập khai báo này, Dagster tự dựng đồ thị tài sản (asset graph) và tự suy ra thứ tự chạy. Bạn không viết DAG task; bạn mô tả dữ liệu tồn tại và quan hệ giữa chúng, orchestrator lo phần còn lại.

Sự đảo ngược trọng tâm này thay đổi mọi thứ:

Thao tác "chạy" một asset gọi là materialize (vật chất hoá): thực thi hàm tính toán để tạo/ghi ra phiên bản mới của tài sản, và Dagster ghi lại materialization event (ai, khi nào, metadata gì — số dòng, min/max, bảng phân vùng).

Dagster trong thực tế: @asset và đồ thị

Trong Dagster, một asset là một hàm Python được đánh dấu @asset. Phụ thuộc được khai báo qua tham số hàm: tham số trùng tên một asset khác nghĩa là "tôi cần asset đó làm đầu vào". Đây là điểm thanh lịch — bạn không nối cạnh thủ công; quan hệ phụ thuộc chính là chữ ký hàm.

# Minh hoạ Dagster — KHÔNG phải SQL chạy được
import pandas as pd
from dagster import asset, AssetCheckResult, asset_check

# Asset nguồn: nạp giao dịch thô từ Core Banking (raw)
@asset(group_name="ingest", compute_kind="python")
def raw_giao_dich(context) -> pd.DataFrame:
    df = doc_tu_core_banking(context.partition_key)  # hàm nghiệp vụ
    context.add_output_metadata({"so_dong": len(df)})
    return df

# Staging: phụ thuộc raw_giao_dich (khai báo qua tham số cùng tên)
@asset(group_name="staging")
def stg_giao_dich(raw_giao_dich: pd.DataFrame) -> pd.DataFrame:
    df = raw_giao_dich.dropna(subset=["account_id", "amount"])
    df["amount"] = df["amount"].astype("float64")
    return df

# Mart gold: đồ thị tự suy ra stg_giao_dich -> mart_du_no
@asset(group_name="mart")
def mart_du_no(stg_giao_dich: pd.DataFrame) -> pd.DataFrame:
    return (stg_giao_dich
            .groupby("account_id", as_index=False)["amount"].sum()
            .rename(columns={"amount": "du_no"}))

Dagster đọc ba hàm trên và tự dựng đồ thị raw_giao_dich → stg_giao_dich → mart_du_no. Không cần một dòng "set_downstream". Khi bạn yêu cầu materialize mart_du_no, Dagster biết phải bảo đảm stg_giao_dich (và raw_giao_dich) đã tươi trước — hoặc materialize chúng theo yêu cầu.

Asset checks: chất lượng dữ liệu gắn vào asset

Một điểm mạnh của mô hình asset: kiểm tra chất lượng (DQ) được gắn trực tiếp vào asset qua asset checks, thay vì là task rời rạc. Một asset check là một khẳng định về tài sản ("mart_du_no không có dư nợ âm"), chạy sau khi materialize và báo kết quả kèm theo asset trong UI.

# Minh hoạ asset check — KHÔNG phải SQL chạy được
@asset_check(asset=mart_du_no, blocking=True)
def khong_co_du_no_am(mart_du_no: pd.DataFrame) -> AssetCheckResult:
    so_am = int((mart_du_no["du_no"] < 0).sum())
    return AssetCheckResult(
        passed=(so_am == 0),
        severity="ERROR",
        metadata={"so_ban_ghi_am": so_am},
    )

Khác với test đứng độc lập, asset check trả lời đúng câu hỏi giải trình: "bảng gold này ở lần cập nhật cuối có đạt DQ không". Với blocking=True, check fail có thể chặn việc materialize các asset hạ nguồn — chính là quality gate kiểu WAP/contract nhưng ở mức từng asset. Đây là cùng triết lý với observability & DRE: tín hiệu chất lượng gắn với đối tượng dữ liệu, không phải với quy trình.

Data-aware scheduling & freshness

Vì orchestrator hiểu dữ liệu, nó lập lịch theo trạng thái dữ liệu chứ không chỉ theo đồng hồ:

• Auto-materialize / declarative automation: bạn khai báo chính sách "materialize mart_du_no mỗi khi stg_giao_dich thượng nguồn có phiên bản mới". Dagster tự phát hiện thay đổi thượng nguồn và kích hoạt — không cần bạn viết sensor thủ công.
• Freshness policy / SLA: gắn vào asset một cam kết như "mart_du_no không được cũ quá 60 phút so với dữ liệu nguồn". Dagster theo dõi và cảnh báo khi asset quá hạn tươi, ngay cả khi mọi task đều "xanh". Đây là bước nhảy chất: SLA đặt trên dữ liệu, không trên job.
• Partitions: asset có thể phân vùng theo ngày (mỗi kỳ EOD là một partition). Materialize một partition, backfill một dải partition — đơn vị vẫn là dữ liệu, nên rất tự nhiên cho báo cáo cuối ngày ngân hàng.

Đồ thị này là thứ Dagster thực sự hiểu và hiển thị — không phải sơ đồ ta vẽ tay để giải thích. Đó là khác biệt cốt lõi.

Tích hợp dbt: mỗi model = một asset

Đây là nơi mô hình asset toả sáng với team đang dùng dbt. Dagster đọc file manifest.json của dbt và biến mỗi dbt model thành một asset Dagster tự động. Lineage của dbt (từ ref()/source()) hợp nhất liền mạch với lineage của các asset Python/nguồn khác thành một đồ thị duy nhất.

Kết quả: bảng gold do dbt build và bảng feature do Python build nằm chung một asset graph, cùng một UI, cùng cơ chế freshness và materialize. Bạn không còn "lineage dbt một chỗ, lineage Airflow một chỗ, ghép tay". dbt test cũng ánh xạ thành asset checks — DQ của dbt hiện ngay cạnh asset trong catalog. Với ngân hàng, đây chính là "single pane of glass" cho lineage từ nguồn thô đến báo cáo giải trình.

Ngoài dbt, Dagster có tích hợp cho các nguồn ingest (dùng embedded ELT như Sling/dlt), kho (Snowflake, BigQuery, Postgres), và ML: một model huấn luyện cũng là một asset, phụ thuộc feature asset → lineage kéo dài đến tận model, thống nhất tư duy với MLOps.

I/O manager: tách logic đọc/ghi khỏi tính toán

Một trừu tượng đắt giá của Dagster là I/O manager. Chú ý ở ví dụ trên: hàm @asset chỉ trả về DataFrame; nó không chứa code ghi vào bảng nào, ở đâu. Việc đọc đầu vào và ghi đầu ra được giao cho I/O manager cấu hình ở tầng deployment.

Lợi ích:

• Cùng một code asset chạy ở nhiều môi trường: dev ghi ra file local/DuckDB, staging và production ghi vào Snowflake — chỉ đổi I/O manager, không sửa logic nghiệp vụ. Điều này khớp với kỷ luật nhiều môi trường mà series đã bàn.
• Đổi storage không đụng logic: chuyển từ Parquet sang bảng warehouse là đổi cấu hình.
• Testable: unit-test hàm asset như hàm Python thuần (truyền DataFrame vào, kiểm tra DataFrame ra), tách bạch khỏi I/O — rất khớp yêu cầu testing của DataOps.

Observability & catalog tích hợp

Dagster đi kèm một UI trong đó asset là công dân hạng nhất: bạn thấy toàn bộ đồ thị asset, trạng thái từng asset (đã materialize lần cuối khi nào, có tươi không, check pass/fail), và metadata từng lần materialize (số dòng, schema, thống kê). Khi kiểm toán hỏi "báo cáo dư nợ hôm qua build từ nguồn nào, lúc mấy giờ, DQ ra sao", câu trả lời nằm ngay trong catalog — không cần dựng công cụ lineage bên ngoài. Đây là điểm khác biệt lớn so với việc ghép Airflow + công cụ lineage + công cụ DQ rời rạc.

So sánh: Dagster vs Airflow vs Prefect

Kim chỉ nam thực dụng: nếu công việc của bạn chủ yếu là tạo ra và duy trì các bảng/dataset/model (đúng như một data platform tự phục vụ), tư duy asset của Dagster ăn khớp tự nhiên. Nếu bạn điều phối đủ loại job không chỉ về dữ liệu (gọi API, gửi mail, batch job hệ thống) và đã đầu tư nặng vào Airflow, ở lại Airflow (và tận dụng Datasets) là hợp lý. Xu hướng chung của ngành đang nghiêng về asset-centric, vì nó phản ánh đúng thứ ta thực sự quan tâm: dữ liệu, không phải quy trình.

Liên hệ tư duy data product / mesh

Không ngẫu nhiên mà "asset" cộng hưởng với khái niệm data product trong Data Mesh. Một data product là một tài sản dữ liệu có chủ sở hữu, có hợp đồng (contract), có SLA về độ tươi và chất lượng — đúng bằng những gì một asset Dagster mang theo: hàm tạo, phụ thuộc, freshness policy, asset checks, metadata. Mô hình hoá kho dữ liệu theo asset là một bước cụ thể để hiện thực hoá tư duy "coi dữ liệu như sản phẩm": mỗi bảng gold không còn là "kết quả phụ của một task" mà là một tài sản được quản trị vòng đời.

Use case thực tế

Bối cảnh: NCB có kho báo cáo giải trình gồm ~120 bảng gold (dư nợ, nợ xấu theo nhóm, huy động, thanh khoản...) phục vụ báo cáo NHNN và ban điều hành. Đang chạy trên Airflow: 6 DAG, lineage nằm rải trong code, DQ là các task test riêng. Sự cố điển hình: dashboard nợ xấu hiện số cũ vì bảng stg_phan_loai_no (ở DAG khác) trễ, nhưng DAG báo cáo vẫn "xanh" và chạy trên dữ liệu cũ — không ai phát hiện đến khi nghiệp vụ khiếu nại.

Chuyển sang mô hình asset (Dagster), mô hình hoá mỗi bảng gold là một asset:

1. Hợp nhất lineage: import toàn bộ dbt project → mỗi model thành asset. mart_no_xau giờ hiển thị rõ chuỗi phụ thuộc đến stg_phan_loai_no và tận nguồn CDC. Phụ thuộc "vô hình xuyên-DAG" trước đây trở nên hiển thị.
2. Freshness policy trên bảng gold trọng yếu: đặt mart_no_xau không được cũ quá 90 phút so với nguồn phân loại nợ. Khi stg_phan_loai_no trễ, mart_no_xau bị đánh dấu quá hạn tươi ngay — cảnh báo bắn trước khi nghiệp vụ nhìn thấy số sai.
3. Asset checks chặn: gắn check blocking "tổng dư nợ mart khớp sổ cái (chênh lệch ≤ 0.01%)" và "không có nhóm nợ ngoài {1..5}". Check đỏ ⇒ không materialize các bảng báo cáo hạ nguồn ⇒ không "đóng dấu" số sai.
4. Auto-materialize theo dữ liệu: mart_no_xau tự build lại mỗi khi thượng nguồn cập nhật, không cần lịch cứng khớp tay giữa các DAG.

Kết quả sau 1 quý: sự cố "báo cáo hiện số cũ" giảm từ ~3 lần/tháng xuống 0 (freshness bắt sớm); thời gian trả lời câu hỏi kiểm toán "bảng này build từ đâu, lúc nào, DQ ra sao" giảm từ ~2 giờ dò log xuống tra trực tiếp trên catalog vài phút; unit-test asset (nhờ tách I/O manager) đưa coverage logic transform lên mức chạy được trong CI mà không cần warehouse thật.

Ghi nhớ

• Task-based (Airflow cổ điển) điều phối công việc; asset-based (Dagster) điều phối dữ liệu. Đảo trọng tâm này cho lineage, freshness và quan sát theo dữ liệu gần như miễn phí.
• Software-Defined Asset: khai báo mỗi bảng/dataset/model là một @asset với hàm tạo và phụ thuộc; phụ thuộc khai báo qua tham số hàm; Dagster tự dựng asset graph và suy ra thứ tự. "Chạy" một asset = materialize.
• Asset checks gắn DQ vào asset (có thể blocking để chặn hạ nguồn) — quality gate mức từng bảng, đúng câu hỏi giải trình.
• Data-aware scheduling: auto-materialize khi thượng nguồn đổi; freshness policy/SLA đặt trên dữ liệu, cảnh báo cả khi task vẫn "xanh".
• Tích hợp dbt: mỗi dbt model → một asset tự động, lineage hợp nhất dbt + Python + ML thành một đồ thị, một catalog.
• I/O manager tách logic đọc/ghi khỏi tính toán ⇒ cùng code chạy dev/staging/prod, dễ đổi storage, dễ unit-test.
• Chọn công cụ: Dagster cho nền tảng dữ liệu asset-centric (lineage/freshness ưu tiên); Airflow cho hệ sinh thái rộng và job đa dạng; Prefect cho workflow Python động, nhẹ. Xu hướng ngành nghiêng về asset-centric.
• Tư duy asset cộng hưởng với data product / Data Mesh: bảng gold là tài sản được quản trị vòng đời, không phải kết quả phụ của một task.