DataOps nâng cao 2 — Xây pipeline ELT thực chiến với dbt

Từ "biết dbt" đến "dựng được pipeline"

Bài dbt 1 — Tổng quan đã trả lời câu hỏi dbt là gì: chữ "T" trong ELT, biên dịch SELECT + Jinja thành CREATE ... AS rồi để warehouse chạy. Bài này không nhắc lại lý thuyết đó. Mục tiêu ở đây là thực chiến: cầm một project dbt trống và biến nó thành một pipeline có tầng, có test, có tài liệu, có lineage — theo đúng cách một team dữ liệu ngân hàng nên làm.

Xuyên suốt bài, ta sẽ dựng một sản phẩm cụ thể: mart "Khách hàng 360" & "Giao dịch" cho NCB, lấy nguồn từ hệ core banking đã được nạp thô vào warehouse (bằng CDC/Fivetran — đó là phần "EL", không phải việc của dbt). Nhiệm vụ của ta là phần "T": làm sạch, chuẩn hoá, tổng hợp, và phơi bày ra các bảng mà bộ phận phân tích rủi ro, marketing, và báo cáo NHNN dùng trực tiếp.

Điểm mấu chốt cần giữ trong đầu suốt bài: dbt biến những câu SQL rời rạc thành một pipeline có cấu trúc. Không còn "file query_final_v3_real.sql" nằm rải rác — mà là một DAG các model phụ thuộc nhau tường minh, mỗi model làm một việc, được test và tài liệu hoá.

Cấu trúc project theo kiến trúc medallion

Cách tổ chức thư mục models/ quyết định độ bền của cả project. Chuẩn được cộng đồng dbt và Databricks phổ biến là medallion (huy chương) — ba tầng chất lượng tăng dần:

models/
├── staging/            -- tầng "bạc thô": 1-1 với nguồn, chỉ làm sạch
│   └── core_banking/
│       ├── _core_banking__sources.yml
│       ├── _core_banking__models.yml
│       ├── stg_core__customers.sql
│       ├── stg_core__accounts.sql
│       └── stg_core__transactions.sql
├── intermediate/       -- tầng "bạc": logic trung gian, tái dùng
│   └── int_transactions_enriched.sql
└── marts/              -- tầng "vàng": bảng nghiệp vụ cuối
    ├── finance/
    │   ├── fct_transactions_monthly.sql
    │   └── dim_customers.sql
    └── _marts__models.yml

Ba tầng có vai trò tách bạch:

Quy tắc vàng: staging là cửa duy nhất chạm vào nguồn thô. Mọi model khác chỉ được ref() tới staging, không bao giờ source() trực tiếp. Nhờ vậy nếu core banking đổi tên cột cust_nm → customer_name, ta chỉ sửa một chỗ trong stg_core__customers, phần còn lại không đụng.

Khai báo sources & kiểm tra freshness

Trước khi làm sạch, phải khai báo nguồn. dbt không tự biết bảng thô nằm ở đâu — ta mô tả trong file YAML. Đây là cấu hình, không phải SQL chạy được:

# models/staging/core_banking/_core_banking__sources.yml
version: 2

sources:
  - name: core_banking            # namespace, dùng source('core_banking', ...)
    database: raw
    schema: core_banking_cdc
    loaded_at_field: _synced_at    # cột do CDC ghi thời điểm nạp
    freshness:                     # ngưỡng "dữ liệu cũ" toàn source
      warn_after:  {count: 6,  period: hour}
      error_after: {count: 24, period: hour}
    tables:
      - name: customers
      - name: accounts
      - name: transactions
        freshness:                 # override riêng cho bảng nóng
          warn_after:  {count: 1, period: hour}
          error_after: {count: 3, period: hour}

loaded_at_field là cột timestamp cho biết bản ghi được nạp lúc nào. Chạy dbt source freshness sẽ so max(_synced_at) với thời điểm hiện tại; nếu vượt ngưỡng thì cảnh báo hoặc lỗi. Với bảng transactions của ngân hàng — nếu CDC tắc 3 tiếng mà không ai biết, mọi báo cáo tính sai — nên đặt ngưỡng chặt hơn hẳn. Việc này thường chạy như một task riêng trong Airflow trước khi build model, để "fail fast" khi nguồn hỏng.

Model đầu tiên: staging chuẩn hoá giao dịch

Staging là nơi "kỷ luật một lần". Một stg_ model điển hình chỉ làm sạch, không hơn: đổi tên cột về snake_case chuẩn, ép kiểu, gom enum vụn thành nhãn nghiệp vụ, lọc bản ghi CDC đã xoá.

Dưới đây là model staging cho giao dịch, viết dưới dạng SELECT chạy được trên sandbox để bạn thấy kết quả thật. (Đây là một model dbt; trong project thật, from transactions sẽ là from {{ source('core_banking', 'transactions') }}.)

-- ▶ Chạy được
-- Model dbt: stg_core__transactions (materialized='view')
-- Trong project thật: FROM {{ source('core_banking','transactions') }}
SELECT
    id                                   AS transaction_id,
    account_id,
    amount                               AS amount_vnd,
    lower(kind)                          AS transaction_kind,
    CASE
        WHEN amount >= 0 THEN 'credit'
        ELSE 'debit'
    END                                  AS direction,
    created_at                           AS transacted_at,
    (created_at)::date                   AS transaction_date
FROM transactions
WHERE amount IS NOT NULL;

Đọc kỹ: một câu SELECT, không JOIN, không GROUP BY. Đó là dấu hiệu của một staging model đúng chuẩn. Mọi logic phức tạp hơn được đẩy lên tầng trên. Về việc chọn view hay table cho tầng này, xem dbt 2 — Models & materializations.

Materializations: chọn đúng "hình hài" cho model

Cùng một câu SELECT, dbt có thể vật chất hoá theo bốn cách. Chọn sai là nguyên nhân số một khiến pipeline chậm hoặc tốn kém:

Nguyên tắc thực dụng cho ngân hàng: staging = view, intermediate = view/ephemeral, dim nhỏ = table, fact giao dịch = incremental.

Fact tổng hợp theo tháng

Mart tài chính cần một bảng fact tổng hợp giao dịch theo tháng và loại tiền — nền cho báo cáo doanh số và biến động dòng tiền. Đây là một fct_ model, viết dạng SELECT chạy được (trong thực tế transactions/accounts sẽ là {{ ref('stg_core__transactions') }}/{{ ref('stg_core__accounts') }}):

-- ▶ Chạy được
-- Model dbt: fct_transactions_monthly (materialized='table' hoặc 'incremental')
-- Thật: JOIN {{ ref('stg_core__transactions') }} + {{ ref('stg_core__accounts') }}
SELECT
    date_trunc('month', t.created_at)::date          AS month,
    a.currency,
    COUNT(*)                                         AS txn_count,
    SUM(t.amount)                                    AS net_flow,
    SUM(CASE WHEN t.amount >= 0 THEN t.amount ELSE 0 END) AS total_credit,
    ROUND(AVG(t.amount)::numeric, 2)                 AS avg_amount
FROM transactions t
JOIN accounts a ON a.id = t.account_id
GROUP BY date_trunc('month', t.created_at), a.currency
ORDER BY month, a.currency;

Lưu ý ép ::numeric trước ROUND — AVG trả về double precision, mà ROUND(double, int) không tồn tại trong PostgreSQL nên phải ép kiểu. Khoá JOIN đúng luồng medallion: transactions.account_id → accounts.id.

Dim khách hàng: JOIN đúng khoá

Chiều dim_customers là xương sống của "Khách hàng 360". Nó gom thông tin định danh khách với chỉ số tài chính tổng hợp. Model dim_ này viết dạng SELECT chạy được (thật ra JOIN các stg_ tương ứng):

-- ▶ Chạy được
-- Model dbt: dim_customers (materialized='table')
-- Thật: các bảng dưới là {{ ref('stg_core__customers') }}, ..._accounts, ..._transactions
SELECT
    c.id                              AS customer_id,
    c.full_name,
    c.city,
    c.created_at                      AS customer_since,
    COUNT(DISTINCT a.id)              AS num_accounts,
    COALESCE(SUM(a.balance), 0)       AS total_balance,
    COUNT(t.id)                       AS lifetime_txn_count,
    ROUND(COALESCE(SUM(t.amount), 0)::numeric, 2) AS lifetime_net_flow
FROM customers c
LEFT JOIN accounts a     ON a.customer_id = c.id
LEFT JOIN transactions t ON t.account_id  = a.id
GROUP BY c.id, c.full_name, c.city, c.created_at
ORDER BY total_balance DESC;

LEFT JOIN để giữ cả khách chưa có tài khoản/giao dịch. Chuỗi khoá đi đúng ba bậc: customers.id ← accounts.customer_id, rồi accounts.id ← transactions.account_id. Vì accounts/transactions không có cột tên khách, ta bắt buộc JOIN về customers mới lấy được full_name.

Incremental models thực chiến

Bảng giao dịch của một ngân hàng có thể hàng tỷ dòng. Build lại table mỗi đêm là không khả thi. Incremental giải quyết: lần đầu build toàn bộ, các lần sau chỉ xử lý bản ghi mới rồi MERGE/INSERT vào bảng đã có. Đây là cấu hình dbt (Jinja), không đánh dấu chạy được:

{{
  config(
    materialized='incremental',
    unique_key='transaction_id',
    incremental_strategy='merge',
    on_schema_change='append_new_columns'
  )
}}

select
    id            as transaction_id,
    account_id,
    amount        as amount_vnd,
    created_at    as transacted_at
from {{ source('core_banking', 'transactions') }}

{% if is_incremental() %}
  -- chỉ lấy bản ghi mới hơn mốc đã có trong bảng đích,
  -- lùi lại 3 ngày để bắt giao dịch "về muộn" (late-arriving)
  where created_at >= (select coalesce(max(transacted_at), '1900-01-01')
                       from {{ this }}) - interval '3 days'
{% endif %}

Ba điểm cốt lõi:

• is_incremental() đúng true chỉ khi bảng đã tồn tại và không chạy --full-refresh. Khối where bên trong nó lọc bản ghi mới; lần đầu (bảng chưa có) khối này bị bỏ qua, dbt quét toàn bộ.
• unique_key + incremental_strategy='merge': nếu một transaction_id đã có, dbt cập nhật thay vì tạo bản trùng — quan trọng khi bản ghi bị chỉnh sửa (ví dụ giao dịch đảo ngược).
• Late-arriving data: giao dịch có thể tới warehouse muộn vài giờ/ngày. Lùi cửa sổ 3 ngày (- interval '3 days') để bắt lại chúng; merge bảo đảm không nhân đôi. Chi tiết tối ưu xem dbt 6 — Incremental & performance.

Snapshots: theo dõi thay đổi với SCD2

Fact/incremental chỉ thêm dòng mới. Nhưng có loại dữ liệu ta cần theo dõi lịch sử thay đổi của cùng một bản ghi — ví dụ hạng khách (VIP → thường), địa chỉ, hạn mức tín dụng. Đây là SCD Type 2 (Slowly Changing Dimension), và dbt giải quyết bằng snapshots:

{% snapshot snap_customers %}
{{
  config(
    target_schema='snapshots',
    unique_key='id',
    strategy='check',
    check_cols=['city', 'segment', 'credit_limit']
  )
}}
select id, full_name, city, segment, credit_limit
from {{ source('core_banking', 'customers') }}
{% endsnapshot %}

Mỗi lần chạy dbt snapshot, dbt so các check_cols với bản mới nhất. Nếu credit_limit của khách thay đổi, dbt "đóng" dòng cũ (điền dbt_valid_to) và chèn dòng mới (dbt_valid_from = now, dbt_valid_to = null). Kết quả là một bảng lịch sử: ta luôn trả lời được "hạn mức của khách X vào ngày Y là bao nhiêu" — thiết yếu cho kiểm toán và mô hình rủi ro. Ngoài strategy='check' còn có strategy='timestamp' (dựa vào cột updated_at của nguồn). Xem dbt 3 — Sources & snapshots.

ref(), source() và DAG tự sinh

Hai hàm Jinja tạo nên sức mạnh của dbt: source() trỏ tới bảng thô đã khai báo, ref() trỏ tới model dbt khác. Bạn không bao giờ viết tên bảng cứng. Nhờ vậy dbt tự suy ra thứ tự phụ thuộc và sinh một DAG — không cần khai báo thứ tự thủ công:

dbt run sẽ build theo đúng topo order của DAG này: staging trước, rồi intermediate, rồi marts — hoàn toàn tự động. Khi ta chạy dbt build, dbt làm cả gói theo thứ tự DAG: chạy model → chạy test của nó → nếu test fail thì dừng nhánh phụ thuộc (không build tiếp mart dựa trên dữ liệu đã sai) → chạy snapshot → nạp seed. Đây là lệnh nên dùng trong production thay vì gọi rời dbt run + dbt test.

Jinja, macros và seeds

Jinja biến SQL tĩnh thành template động. Khi một đoạn logic lặp lại nhiều model, gói nó vào macro (như hàm):

-- macros/vnd_to_usd.sql
{% macro vnd_to_usd(amount_col, rate=24500) %}
    round( ({{ amount_col }})::numeric / {{ rate }}, 2)
{% endmacro %}

-- dùng trong model:
select {{ vnd_to_usd('amount_vnd') }} as amount_usd from ...

Macro giúp chuẩn hoá logic tiền tệ/masking PII một chỗ, tránh copy-paste. Chi tiết ở dbt 5 — Jinja & macros.

Seeds là các file CSV nhỏ trong seeds/ được dbt seed nạp thành bảng — hợp cho dữ liệu tra cứu tĩnh: bảng mã chi nhánh, mã tiền tệ ISO, danh mục ngành nghề. Không dùng seed cho dữ liệu lớn hay hay đổi.

Tài liệu, style guide và cấu hình môi trường

Tài liệu inline: mô tả (description) đặt ngay trong YAML của model/cột. dbt docs generate gom chúng thành một site tài liệu có cả sơ đồ lineage click được:

# models/marts/finance/_marts__models.yml
models:
  - name: dim_customers
    description: "Chiều khách hàng cho Khách hàng 360. Hạt: 1 dòng/khách."
    columns:
      - name: customer_id
        description: "Khoá khách hàng, duy nhất."
        tests: [unique, not_null]
      - name: total_balance
        description: "Tổng số dư mọi tài khoản (VND)."

Quy ước đặt tên (style guide) cần thống nhất toàn team, nếu không project sẽ hỗn loạn khi có nhiều người:

• Model: stg_<nguồn>__<thực_thể>, int_<mô_tả>, fct_<domain> / dim_<thực_thể>.
• Cột: snake_case; khoá kết bằng _id; boolean bắt đầu is_/has_; timestamp kết _at, date kết _date.
• Mỗi model một trách nhiệm; SQL dùng CTE đặt tên rõ, không subquery lồng sâu.

Cấu hình theo môi trường: file profiles.yml khai báo nhiều target (dev/prod) trỏ tới schema/credential khác nhau. Dev build vào dbt_kienpv (schema cá nhân), prod build vào analytics. Chuyển bằng dbt run --target prod. Nhờ đó dev thử nghiệm không đụng bảng production — nền tảng cho CI/CD analytics.

Use case thực tế

Bối cảnh: NCB cần dựng mart "Khách hàng 360" phục vụ đội rủi ro tín dụng, từ nguồn core banking đã nạp qua CDC. Bảng transactions có ~1,8 tỷ dòng, tăng ~4 triệu dòng/ngày.

Cách làm theo medallion:

1. Sources: khai báo core_banking với transactions freshness error_after: 3h. Airflow chạy dbt source freshness lúc 1h sáng; nếu CDC trễ, báo động Slack, dừng pipeline — tránh báo cáo sai.
2. Staging: stg_core__customers/accounts/transactions (view) chuẩn hoá tên cột, ép kiểu, chuyển timezone về Asia/Ho_Chi_Minh.
3. Incremental: fct_transactions dùng merge + unique_key, cửa sổ late-arriving 3 ngày. Kết quả: build đêm chỉ xử lý ~12 triệu dòng (3 ngày) thay vì 1,8 tỷ — thời gian build từ ~40 phút xuống ~90 giây.
4. Snapshot: snap_customers theo dõi segment và credit_limit (SCD2) cho mô hình rủi ro và kiểm toán.
5. Marts: dim_customers + fct_transactions_monthly, tài liệu hoá đầy đủ, test unique/not_null trên khoá.

Kết quả: một dbt build duy nhất (chạy qua Airflow) dựng lại toàn bộ mart, chạy 40+ test, sinh tài liệu + lineage. Đội rủi ro query thẳng dim_customers thay vì viết SQL 500 dòng lên bảng thô. Khi core đổi tên cột, chỉ sửa một staging model.

Ghi nhớ

• Medallion: staging (stg_, 1-1 làm sạch, không JOIN) → intermediate (int_, logic tái dùng) → marts (fct_/dim_ theo domain). Staging là cửa duy nhất chạm nguồn thô.
• Sources + freshness: khai báo nguồn trong YAML, đặt ngưỡng chặt cho bảng nóng (giao dịch), dbt source freshness để fail-fast.
• Materialization: view cho staging/intermediate, table cho dim nhỏ, incremental cho fact giao dịch lớn, ephemeral cho bước trung gian nhỏ.
• Incremental: is_incremental() + unique_key + strategy='merge'; lùi cửa sổ vài ngày để bắt late-arriving mà không nhân đôi nhờ merge.
• Snapshots: SCD2 để theo dõi lịch sử bản ghi (hạn mức, hạng khách) — thiết yếu cho audit và rủi ro.
• ref()/source() sinh DAG tự động; dbt build chạy model + test + snapshot + seed theo thứ tự, dừng nhánh khi test fail.
• Jinja/macros gói logic lặp; seeds cho dữ liệu tra cứu tĩnh; description → docs + lineage; target dev/prod tách môi trường.
• dbt biến SQL rời rạc thành pipeline có cấu trúc, test, tài liệu, lineage — không còn query mồ côi.