DataOps nâng cao 10 — Pipeline điều khiển bằng metadata

Vấn đề: không thể code tay từng pipeline khi có hàng trăm bảng

Các bài trước của series dạy cách làm tốt một pipeline: dbt cho ELT, Airflow điều phối, CI/CD, data contract, observability, rồi platform tự phục vụ để làm tốt hàng loạt. Bài này giải quyết mặt còn lại của quy mô: khi NCB cần đưa hàng trăm bảng từ core banking, card switch, hệ thống thẻ, LOS (loan origination), CRM… vào lakehouse, ta không thể ngồi viết tay từng pipeline một.

Hãy làm phép tính. Giả sử core banking có 400 bảng cần nạp. Nếu mỗi bảng cần một pipeline riêng — một DAG Airflow, một loader, một model staging dbt, vài test — và mỗi cái tốn nửa ngày, thì ta mất 200 ngày-người chỉ để onboarding lần đầu. Tệ hơn: 400 pipeline gần-như-giống-nhau đó phải được bảo trì. Khi core đổi kiểu cột ACCT_BAL, hoặc khi ta muốn thêm một cột audit vào mọi bảng landing, ta phải sửa 400 chỗ. Đây chính là vi phạm DRY (Don't Repeat Yourself) ở tầng pipeline: cùng một logic lặp lại 400 lần, chỉ khác tên bảng và mapping cột.

Lời giải là metadata-driven pipeline: thay vì viết pipeline, ta viết một framework đọc metadata/config (danh sách bảng nguồn, mapping cột, quy tắc biến đổi, lịch, test cần áp) rồi sinh ra pipeline/DAG động. Onboarding một bảng mới trở thành thêm một dòng config, không phải viết code. Logic biến đổi chung sửa một lần trong framework, áp cho tất cả.

Metadata-driven là gì

Nguyên tắc cốt lõi: tách "cái gì" khỏi "làm thế nào".

• Metadata (khai báo, dữ liệu về pipeline): bảng CIF.CUSTOMER ở core, nạp vào lakehouse.raw.customer, tần suất hàng ngày lúc 2h, khoá chính CUST_ID, cột PHONE/ID_NUMBER là PII cần mask, chạy test not_null trên khoá. Đây là cấu hình, thường sống trong YAML hoặc một bảng metadata trong database.
• Engine/framework (thực thi, "làm thế nào"): một khối code duy nhất biết cách đọc config đó và biến nó thành pipeline chạy được — kết nối nguồn, extract, landing, sinh model, sinh test, đăng ký lineage.

Điểm mấu chốt: thêm nguồn mới không cần đụng code engine. Engineer khai báo bảng mới trong YAML, mở PR, review config (không review logic), merge — pipeline tự xuất hiện. Đây là cách một đội nhỏ vận hành hàng trăm pipeline.

Config-driven ingestion

Tầng dễ metadata-hoá nhất là ingestion (Extract-Load), vì nó lặp lại nhiều nhất: đọc từ nguồn, ghi vào landing zone. Thay vì viết loader cho từng bảng, ta định nghĩa một spec khai báo cho mỗi nguồn:

# metadata/sources/core_banking.yml — MINH HOẠ (không phải code chạy)
source: core_banking
connection: "{{ conn.core_oracle }}"
default_schedule: "0 2 * * *"        # 2h sáng hàng ngày
target_layer: raw
tables:
  - name: CUSTOMER
    target: raw.customer
    load_mode: cdc                    # full | incremental | cdc
    primary_key: [CUST_ID]
    watermark: LAST_UPD_TS
    pii_columns: [PHONE, ID_NUMBER, EMAIL]
    tests: [not_null_pk, unique_pk, freshness_24h]
  - name: ACCOUNT
    target: raw.account
    load_mode: incremental
    primary_key: [ACCT_ID]
    watermark: LAST_UPD_TS
    tests: [not_null_pk, freshness_24h]
  - name: TRANSACTION
    target: raw.transaction
    load_mode: incremental
    primary_key: [TXN_ID]
    watermark: TXN_TS
    partition_by: TXN_DATE
    tests: [not_null_pk, freshness_6h]

Engine ingestion đọc file này và với mỗi entry trong tables, nó biết đủ để tự nạp: dùng connection nào, extract full/incremental/CDC ra sao (dựa load_mode và watermark), khử trùng lặp theo primary_key, ghi vào target, mask cột trong pii_columns, và đăng ký test. Không cần viết một dòng code cho bảng TRANSACTION — chỉ khai báo nó.

Khi core thêm bảng STANDING_ORDER, engineer thêm một khối - name: STANDING_ORDER .... Đó là toàn bộ công việc. Khi muốn thêm cột _loaded_at vào mọi bảng raw, ta sửa engine một lần. So sánh: nếu code tay, đó là 400 sửa đổi.

Dynamic DAG generation trong Airflow

Airflow sinh DAG bằng cách thực thi file Python trong thư mục dags/. File đó không nhất thiết phải hard-code một DAG — nó có thể đọc config và tạo DAG trong vòng lặp. Đây là "dynamic DAG generation": một file Python sinh ra nhiều DAG/task từ metadata.

# dags/generate_ingestion_dags.py — PSEUDOCODE minh hoạ, không phải mã production
import yaml
from pathlib import Path
from airflow import DAG
from airflow.decorators import task

for cfg_file in Path("metadata/sources").glob("*.yml"):
    spec = yaml.safe_load(cfg_file.read_text())
    source = spec["source"]

    # Một DAG cho mỗi nguồn; mỗi bảng là một task trong DAG
    dag = DAG(
        dag_id=f"ingest_{source}",
        schedule=spec["default_schedule"],
        catchup=False,
        tags=["metadata-driven", source],
    )
    with dag:
        for tbl in spec["tables"]:
            @task(task_id=f"load_{tbl['name'].lower()}")
            def load(cfg=tbl, conn=spec["connection"]):
                # Engine dùng chung: extract theo load_mode, landing, mask PII
                run_ingestion(conn=conn, table_spec=cfg)
            load()

    globals()[dag.dag_id] = dag   # Airflow nhặt DAG từ globals()

Vòng lặp trên đọc mọi file YAML nguồn và sinh một DAG ingest_core_banking, ingest_card_switch… mỗi DAG có N task tương ứng N bảng. Thêm một nguồn = thêm một file YAML; Airflow tự có thêm DAG ở lần parse tiếp theo. Về mô hình task/TaskFlow và truyền dữ liệu giữa task, xem Airflow 4 — TaskFlow & XCom.

Lưu ý vận hành quan trọng: scheduler của Airflow parse lại các file DAG rất thường xuyên (mặc định vài chục giây/lần). Nếu file generator đi kết nối database hay gọi API mỗi lần parse để lấy config, ta tạo tải nặng vô hình và làm scheduler chậm. Nguyên tắc: parse phải nhẹ. Đọc YAML tĩnh trên disk thì ổn; nếu metadata nằm trong bảng database, hãy snapshot ra file (render config thành YAML/JSON trong bước build/CI) rồi để generator đọc file, đừng query live trong lúc parse.

Có hai kiểu sinh, cần phân biệt:

Với quy mô ngân hàng và yêu cầu kiểm soát thay đổi, rendered/build-time thường an toàn hơn: mỗi DAG là một artifact được review và có dấu vết trong Git.

dbt tận dụng metadata

dbt vốn đã là công cụ metadata-driven: mỗi model có schema.yml khai báo cột, test, mô tả. Nhưng ta có thể đẩy xa hơn — sinh chính bản thân model từ metadata thay vì viết tay từng file .sql.

Codegen từ metadata: gói dbt-codegen (macro generate_source, generate_base_model) sinh khai báo source và model staging từ cấu trúc bảng thực tế trong warehouse, thay vì gõ tay hàng trăm cột. Với 400 bảng landing, đây là khác biệt giữa vài phút và vài tuần.

Macro sinh SQL từ config: dùng Jinja để một model staging đọc mapping cột từ biến/seed và tự dựng phần SELECT. Ví dụ, một macro stage_from_config nhận tên bảng và danh sách mapping, sinh ra các dòng raw_col AS clean_col:

-- models/staging/stg_account.sql — MINH HOẠ dbt/Jinja, KHÔNG chạy trong sandbox
{{ config(materialized='view') }}
{% set mapping = var('stg_account_mapping') %}
select
    {%- for m in mapping %}
    {{ m.raw }} as {{ m.clean }}{{ "," if not loop.last }}
    {%- endfor %}
from {{ source('raw', 'account') }}

dbt-utils cũng cho các macro tạo SQL động (ví dụ star, get_column_values, union_relations) để tránh liệt kê cột thủ công. Về cách macro và Jinja hoạt động, và giới hạn nên đặt ra, xem dbt 5 — Jinja & macro.

Điểm cân bằng: dbt-codegen sinh một lần rồi ta commit file .sql sinh ra như code bình thường (dễ đọc, review được) là cách lành mạnh nhất. Sinh SQL hoàn toàn động lúc chạy bằng Jinja lồng nhiều tầng thì mạnh nhưng nhanh chóng thành thứ không ai đọc nổi — dùng có chừng mực.

Metadata cho test và contract

Chất lượng cũng metadata-hoá được. Thay vì viết test cho từng bảng, ta khai báo bộ test theo tên trong metadata rồi áp hàng loạt:

• Trong ví dụ YAML ở trên, mỗi bảng có tests: [not_null_pk, freshness_24h]. Engine dịch tên đó thành test dbt cụ thể (not_null trên cột trong primary_key, freshness dựa cột watermark). Muốn siết freshness cho toàn bộ nguồn giao dịch từ 24h xuống 6h? Sửa một chỗ trong định nghĩa freshness_6h hoặc trong config, không sửa 400 file.
• Data contract cũng là metadata: schema kỳ vọng, kiểu cột, ràng buộc, chủ sở hữu — khai báo một lần, framework kiểm tra mọi bảng tuân theo. Kế thừa trực tiếp DataOps nâng cao 5 — Quality & contracts: metadata-driven là cách áp contract ở quy mô hàng trăm bảng mà không viết tay từng cái.

Kết quả: một chính sách chất lượng thống nhất (mọi khoá chính phải not_null + unique, mọi cột PII phải được mask, mọi bảng phải có freshness) được thực thi tự động trên toàn bộ danh mục, chỉ nhờ khai báo.

Metadata store/registry & lineage tự sinh

Khi config là nguồn chân lý, nó nên sống ở một chỗ tra cứu được: metadata registry. Có thể là các file YAML trong Git (đơn giản, review qua PR, có lịch sử), hoặc một schema bảng trong database (query được bằng SQL, phù hợp khi có UI quản trị). Nhiều đội dùng cả hai: YAML là nguồn chân lý, được đồng bộ vào bảng để tra cứu.

Lợi ích lớn của cách tiếp cận này là lineage gần như miễn phí. Vì engine biết mọi bảng đọc từ đâu và ghi vào đâu (nó vừa dùng chính thông tin đó để sinh pipeline), nó có thể tự phát ra lineage vào catalog — không cần khai tay. Mỗi lần sinh pipeline, engine đăng ký: raw.account đến từ core_banking.ACCOUNT, staging.stg_account đến từ raw.account. Về catalog và lineage, xem Governance — Catalog & lineage.

Truy vấn "metadata" bằng SQL — minh hoạ

Để cảm nhận việc "coi cấu hình như dữ liệu để truy vấn", đây là một truy vấn chạy được trên sandbox: ta dùng chính 5 bảng sandbox như thể chúng là danh mục đối tượng cần onboard, và đếm/liệt kê chúng — đúng kiểu một metadata registry trả lời "có bao nhiêu đối tượng theo từng nhóm, khối lượng bao nhiêu":

-- ▶ Chạy được
SELECT c.city AS domain,
       COUNT(DISTINCT a.id)  AS so_tai_khoan,
       COUNT(t.id)           AS so_ban_ghi,
       COUNT(DISTINCT e.department_id) AS so_don_vi
FROM customers c
LEFT JOIN accounts a     ON a.customer_id = c.id
LEFT JOIN transactions t ON t.account_id = a.id
LEFT JOIN employees e    ON e.department_id IN (SELECT id FROM departments)
GROUP BY c.city
ORDER BY so_ban_ghi DESC;

Ý tưởng chuyển vị: nếu thay 5 bảng này bằng bảng metadata.tables(source, target, load_mode, pii_flag, freshness_sla), thì cùng một kiểu truy vấn cho ta báo cáo vận hành metadata — bao nhiêu bảng theo từng nguồn, bao nhiêu bảng CDC vs full, bao nhiêu bảng còn thiếu test. Config trở thành thứ query được, không chỉ là file cấu hình câm.

Templating & scaffolding

Metadata-driven bổ sung cho scaffolding đã nói ở DataOps nâng cao 7 — Platform & self-service. Golden path của platform cung cấp khung; metadata điền nội dung vào khung đó. Một domain mới không copy-paste pipeline; họ khai báo nguồn của mình vào registry và framework của platform lo phần còn lại. Đây là cách self-service và metadata-driven khớp nhau: platform sở hữu engine, domain sở hữu config.

Lợi ích và rủi ro — cân bằng framework-hoá với linh hoạt

Nguyên tắc cân bằng thực dụng:

• Framework hoá cái lặp lại, chừa cửa cho cái đặc thù. 90% bảng vừa khuôn chuẩn — để config lo. 10% bảng đặc thù (join phức tạp, logic nghiệp vụ riêng) — cho phép viết model dbt tay, đừng ép vào framework.
• Framework là code sản phẩm, phải test. Engine sinh 400 pipeline thì bug của nó nhân lên 400 lần. Nó cần unit test, cần CI, cần versioning y như bất kỳ code nào — có phần trong DataOps nâng cao 4 — CI/CD analytics.
• Ưu tiên "sinh rồi commit" hơn "sinh lúc chạy". Khi được, hãy render config thành artifact (DAG tĩnh, model .sql) để review và debug được, thay vì mọi thứ vô hình sinh runtime.
• Đầu tư tài liệu và training. Metadata-driven chỉ có giá trị nếu cả đội đọc được config và hiểu engine. Một framework không ai ngoài tác giả hiểu là nợ kỹ thuật, không phải tài sản.

Use case thực tế

Bối cảnh: NCB cần onboarding 380 bảng từ core banking (Oracle) và card switch vào lakehouse trong quý, để phục vụ báo cáo rủi ro và data mart cho các domain. Đội data platform chỉ có 4 người — không thể viết tay 380 pipeline.

Cách làm metadata-driven:

1. Sinh config ban đầu tự động. Dùng dbt-codegen/query metadata Oracle để xuất danh sách bảng, cột, kiểu, khoá chính ra YAML source spec — 380 khối config sinh sẵn, engineer chỉ review và bổ sung load_mode, cột PII, SLA freshness. Việc này mất ~1 tuần thay vì hàng tháng gõ tay.
2. Một engine ingestion đọc YAML, sinh DAG Airflow theo nguồn (rendered ở CI, không parse-time): ingest_core_banking với ~300 task, ingest_card_switch với ~80 task.
3. Chuẩn hoá & mask đồng loạt: engine tự thêm cột audit _loaded_at, _source, và mask 42 cột PII (số CMND, số thẻ, SĐT, email) theo khai báo pii_columns — không đội nào tự nhớ phải mask.
4. Test hàng loạt: mỗi bảng nhận bộ test theo tên trong config (not_null + unique khoá chính, freshness). Tổng ~1.100 test dbt được khai báo, không viết.
5. Lineage tự sinh: engine đăng ký lineage Oracle → raw → staging vào catalog cho cả 380 bảng khi sinh pipeline.

Kết quả minh hoạ: onboarding 380 bảng trong ~3 tuần với 4 người, thay vì ~190 ngày-người ước tính nếu code tay. Sau đó, thêm một bảng mới = một PR sửa YAML, review trong 15 phút. Khi audit yêu cầu mask thêm một loại cột nhạy cảm trên toàn bộ, đội sửa một quy tắc trong engine và chạy lại — áp cho cả 380 bảng, thay vì mở 380 vé sửa tay. Cái giá: đội phải viết tài liệu engine và 2 người phải hiểu sâu framework để không thành hộp đen; 10% bảng có logic đặc thù (ví dụ bảng giao dịch hợp nhất nhiều hệ) vẫn được viết tay ngoài framework.

Ghi nhớ

• Vấn đề của quy mô: hàng trăm bảng gần-giống-nhau → không code tay từng cái. Metadata-driven là áp DRY ở tầng pipeline.
• Tách "cái gì" (metadata/config) khỏi "làm thế nào" (engine). Onboard nguồn mới = thêm config, không viết code; sửa chính sách = sửa engine một lần, áp tất cả.
• Config-driven ingestion: khai báo nguồn/đích/load_mode/test/PII trong YAML hoặc bảng metadata; engine tự nạp.
• Dynamic DAG generation trong Airflow: file Python đọc config sinh DAG trong vòng lặp. Giữ parse nhẹ (đọc file tĩnh, đừng query live). Ưu tiên rendered/build-time để DAG review được trong Git.
• dbt metadata-driven: dbt-codegen sinh source/model, macro Jinja + dbt-utils sinh SQL từ config. Sinh rồi commit lành mạnh hơn sinh runtime lồng nhiều tầng.
• Test & contract khai báo hàng loạt: bộ test theo tên áp cho toàn danh mục; một chính sách chất lượng, thực thi tự động.
• Metadata registry (YAML/Git hoặc bảng DB) làm nguồn chân lý; lineage tự sinh vào catalog vì engine đã biết luồng dữ liệu.
• Rủi ro: abstraction quá mức, hộp đen, framework khó debug — cần test cho chính engine, tài liệu tốt, và chừa đường thoát cho 10% nguồn đặc thù.