DataOps nâng cao 17 — Dựng dự án dbt + Airflow từ 0

Mục tiêu bài này: ghép mọi thứ đã học thành một dự án chạy được

Mười sáu bài trước của series đã dạy từng mảnh: kiến trúc lakehouse, dbt để transform, Airflow để orchestrate, CI/CD, data quality. Nhưng đọc từng mảnh rời rạc thì khó hình dung chúng khớp vào nhau ra sao. Bài này là bài hands-on: ta cùng dựng một dự án DataOps hoàn chỉnh từ số 0 tới production, cho một kịch bản ngân hàng cụ thể là pipeline "Khách hàng 360 & Giao dịch".

Đầu ra cuối cùng: hai data mart phục vụ báo cáo/BI — fct_transactions_monthly (giao dịch tổng hợp theo tháng, theo tiền tệ) và dim_customer (chiều khách hàng để join/lọc). Hai mart này build tự động mỗi đêm bằng Airflow, kiểm chất lượng bằng dbt test, và mọi thay đổi phải qua CI trên Pull Request. Đây là xương sống thực tế để một team dữ liệu ngân hàng bắt đầu — không phải đồ chơi, mà đúng khuôn để mở rộng.

Bài này giả định bạn đã nắm nền tảng từ dbt ELT, Airflow orchestration, CI/CD analytics và quality & contracts. Ở đây ta không giảng lại lý thuyết, ta ráp chúng.

Bức tranh tổng thể

Luồng làm việc: dev viết model trên máy → mở PR → CI chạy dbt build+test trên schema tạm → merge vào main → Airflow chạy pipeline production hằng đêm (EL → run → test → publish) → mart sẵn sàng cho BI. Ta sẽ dựng lần lượt từng khối.

Bước 1 — Khởi tạo repo & cấu trúc thư mục

Nguyên tắc: một repo duy nhất chứa cả dbt project, các DAG Airflow, tests và cấu hình CI. Mono-repo giúp mọi thay đổi transform + orchestration + quality đi chung một Pull Request, review một lần, deploy đồng bộ.

kh360-dataops/
├── dbt_project.yml            # cấu hình dbt project
├── profiles/                  # profiles theo môi trường (không commit secret thật)
│   └── profiles.yml
├── models/
│   ├── staging/               # làm sạch 1-1 từ source
│   │   ├── _sources.yml        # khai báo SOURCES + freshness
│   │   ├── _staging.yml        # schema tests cho staging
│   │   ├── stg_customers.sql
│   │   ├── stg_accounts.sql
│   │   └── stg_transactions.sql
│   ├── intermediate/          # bước trung gian, không expose ra BI
│   │   └── int_transactions_enriched.sql
│   └── marts/                 # bảng phục vụ báo cáo/BI
│       ├── _marts.yml
│       ├── fct_transactions_monthly.sql
│       └── dim_customer.sql
├── tests/                     # singular tests (SQL đối soát tuỳ biến)
│   └── assert_monthly_totals_reconcile.sql
├── dags/
│   └── kh360_daily.py         # Airflow DAG production
├── .github/workflows/
│   └── ci.yml                 # CI trên Pull Request
└── README.md

Ba lớp model theo mô hình medallion/layered: staging (bronze-ish, làm sạch 1-1), intermediate (silver, ghép/enrich), marts (gold, phục vụ tiêu dùng). Quy ước đặt tên rõ ràng: stg_, int_, fct_/dim_. Đây là quy ước dbt phổ biến, giúp người mới đọc repo hiểu ngay vai trò từng file.

Bước 2 — Cấu hình kết nối & profiles theo môi trường

Nguyên tắc vàng: không bao giờ hardcode mật khẩu/host vào file commit. profiles.yml chỉ chứa tên biến môi trường, giá trị thật được inject lúc chạy (từ secret manager của ngân hàng, biến CI, hay Airflow Connection). Ta tách rõ target dev và prod.

# profiles/profiles.yml — KHÔNG commit secret, chỉ đọc từ env
kh360:
  target: dev
  outputs:
    dev:
      type: postgres
      host: "{{ env_var('DBT_HOST') }}"
      user: "{{ env_var('DBT_USER') }}"
      password: "{{ env_var('DBT_PASSWORD') }}"
      port: 5432
      dbname: warehouse
      schema: dev_{{ env_var('DBT_DEV_USER', 'analyst') }}   # mỗi dev một schema riêng
      threads: 4
    prod:
      type: postgres
      host: "{{ env_var('DBT_HOST') }}"
      user: "{{ env_var('DBT_PROD_USER') }}"
      password: "{{ env_var('DBT_PROD_PASSWORD') }}"
      port: 5432
      dbname: warehouse
      schema: analytics          # schema production cố định
      threads: 8

Điểm cần nhớ: mỗi dev ghi vào schema riêng (dev_binh, dev_lan...) nên không giẫm chân nhau; prod ghi vào analytics. Chọn target bằng dbt run --target prod. Cách tách môi trường này khớp với chiến lược environments/WAP đã bàn trong series.

Bước 3 — Khai báo SOURCES + source freshness

Trước khi transform, dbt cần biết dữ liệu thô ("raw") nằm ở đâu. Ta khai báo trong _sources.yml. Với ngân hàng, ba nguồn cốt lõi là customers, accounts, transactions. Ta cũng bật freshness — nếu bảng transactions không có bản ghi mới trong X giờ, coi như nguồn "ôi", cảnh báo trước khi build ra số sai.

# models/staging/_sources.yml
version: 2
sources:
  - name: core_banking
    database: warehouse
    schema: raw
    tables:
      - name: customers
        columns:
          - name: id
      - name: accounts
      - name: transactions
        loaded_at_field: created_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}

Chạy dbt source freshness sẽ báo lỗi nếu giao dịch mới nhất cũ quá 24 giờ — dấu hiệu job nạp dữ liệu (EL) đã hỏng. Chi tiết về sources/freshness xem dbt sources & snapshots.

Bước 4 — Viết STAGING models (làm sạch, chuẩn hoá)

Staging là lớp 1-1 với source: đổi tên cột về snake_case chuẩn, ép kiểu, chuẩn hoá giá trị (ví dụ upper-case currency, chuẩn hoá kind). Không join, không tổng hợp ở đây — mỗi staging model tương ứng đúng một source table.

Đây là staging cho giao dịch. Trên sandbox học tập của app này, raw.transactions chính là bảng transactions, nên block dưới chạy được:

-- ▶ Chạy được
WITH src AS (
  SELECT id, account_id, amount, kind, created_at
  FROM transactions
)
SELECT
  id                                   AS transaction_id,
  account_id,
  amount::numeric                      AS amount,
  LOWER(TRIM(kind))                    AS kind,
  created_at,
  CAST(created_at AS DATE)             AS transaction_date
FROM src;

Trong dbt thật, file stg_transactions.sql dùng FROM {{ source('core_banking', 'transactions') }} thay cho FROM transactions, còn logic làm sạch giữ nguyên. Tương tự có stg_accounts (chuẩn hoá currency viết hoa, ép balance::numeric) và stg_customers (chuẩn full_name, city). Ví dụ stg_accounts chạy được trên sandbox:

-- ▶ Chạy được
SELECT
  id                       AS account_id,
  customer_id,
  account_no,
  balance::numeric         AS balance,
  UPPER(currency)          AS currency
FROM accounts;

Bước 5 — INTERMEDIATE + MARTS (SELECT thật)

Intermediate ghép các staging lại nhưng chưa expose ra BI. Ví dụ int_transactions_enriched gắn mỗi giao dịch với tiền tệ và khách hàng của tài khoản (join transactions → accounts → customers). Đây là chỗ khớp đúng khoá join.

Mart 1 — fct_transactions_monthly: bảng fact tổng hợp giao dịch theo tháng × tiền tệ, phục vụ báo cáo BI. Trên sandbox, ta viết trực tiếp từ các bảng gốc (thay cho ref staging), block này chạy được:

-- ▶ Chạy được
SELECT
  DATE_TRUNC('month', t.created_at)::date        AS month,
  a.currency,
  COUNT(*)                                        AS txn_count,
  ROUND(SUM(t.amount)::numeric, 2)                AS total_amount,
  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;

Mart 2 — dim_customer: chiều khách hàng, ghép thông tin khách với chỉ số tài khoản (số tài khoản, tổng số dư quy theo khách). Vì accounts/transactions không có cột tên khách, ta phải JOIN sang customers để lấy full_name. Block dưới chạy được — join đúng khoá accounts.customer_id → customers.id:

-- ▶ Chạy được
SELECT
  c.id                                    AS customer_id,
  c.full_name,
  c.city,
  c.created_at                            AS customer_since,
  COUNT(a.id)                             AS num_accounts,
  ROUND(COALESCE(SUM(a.balance), 0)::numeric, 2) AS total_balance
FROM customers c
LEFT JOIN accounts a ON a.customer_id = c.id
GROUP BY c.id, c.full_name, c.city, c.created_at
ORDER BY total_balance DESC;

Trong repo dbt, hai file này là fct_transactions_monthly.sql và dim_customer.sql, dùng {{ ref('stg_transactions') }}, {{ ref('stg_accounts') }}, {{ ref('stg_customers') }} thay vì tên bảng thô. ref() chính là thứ giúp dbt tự dựng thứ tự build (lineage) — không cần ta viết tay thứ tự chạy.

Bước 6 — Thêm TESTS (schema tests + 1 singular test)

Model không có test là model không đáng tin. dbt có hai loại: generic test khai báo trong schema.yml (not_null, unique, relationships, accepted_values) và singular test là một câu SQL tuỳ biến — trả về 0 dòng nghĩa là PASS, trả về dòng nào là FAIL.

Khai báo generic tests cho marts:

# models/marts/_marts.yml
version: 2
models:
  - name: dim_customer
    columns:
      - name: customer_id
        tests: [not_null, unique]        # mỗi khách xuất hiện đúng 1 lần
      - name: total_balance
        tests:
          - not_null
  - name: fct_transactions_monthly
    columns:
      - name: currency
        tests:
          - not_null
          - accepted_values:
              values: ['USD', 'VND', 'EUR']
    # kiểm quan hệ: currency phải tồn tại trong dim (minh hoạ relationships)

relationships là test tham chiếu khoá ngoại: ví dụ trong int_transactions_enriched, cột customer_id phải tồn tại trong dim_customer.customer_id — bắt được giao dịch "mồ côi" trỏ về khách không có thật.

Singular test — đối soát tổng: tổng total_amount trong mart tháng phải khớp tổng amount thô. Nếu lệch, có model tính sai. Trên sandbox, logic đối soát này chạy được (trả 0 dòng = khớp = PASS):

-- ▶ Chạy được
WITH mart AS (
  SELECT ROUND(SUM(total_amount)::numeric, 2) AS s
  FROM (
    SELECT SUM(t.amount) AS total_amount
    FROM transactions t
    JOIN accounts a ON a.id = t.account_id
    GROUP BY DATE_TRUNC('month', t.created_at), a.currency
  ) m
),
raw AS (
  SELECT ROUND(SUM(amount)::numeric, 2) AS s FROM transactions
)
SELECT mart.s AS mart_total, raw.s AS raw_total
FROM mart, raw
WHERE mart.s <> raw.s;

Trong dbt, file tests/assert_monthly_totals_reconcile.sql chứa đúng câu này (dùng ref()); dbt tự coi nó là test và fail nếu có dòng trả về. Loại test đối soát tổng này là "vũ khí" quan trọng nhất trong ngân hàng — bắt lỗi rơi số/nhân đôi trước khi số lên báo cáo. Xem thêm quality & contracts và dbt tests & docs.

Bước 7 — dbt docs & lineage

dbt docs generate && dbt docs serve sinh trang tài liệu tự động: mô tả từng model/cột (lấy từ description trong yml), cùng lineage graph dựng từ mạng lưới ref()/source(). Với ngân hàng, lineage là "vàng" cho audit: khi kiểm toán hỏi "số này từ đâu ra?", ta chỉ vào đồ thị từ raw.transactions → stg_transactions → fct_transactions_monthly → báo cáo. Không cần vẽ tay, không lệch với code.

Lineage dự án của ta:

Bước 8 — Đóng gói vào Airflow DAG

Đến giờ ta chạy dbt bằng tay. Production cần tự động mỗi đêm. Airflow DAG điều phối bốn bước tuần tự: EL (nạp raw từ core banking) → dbt run (build model) → dbt test (kiểm chất lượng) → publish (docs/thông báo). Nguyên tắc quan trọng: test fail thì DAG fail — không để số xấu lặng lẽ lên báo cáo.

# dags/kh360_daily.py  (minh hoạ — không đánh dấu chạy được)
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

DBT = "cd /opt/kh360-dataops && dbt"

with DAG(
    dag_id="kh360_daily",
    schedule="0 2 * * *",          # 02:00 mỗi ngày
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={"retries": 2},
    tags=["kh360", "marts"],
) as dag:

    el = BashOperator(
        task_id="extract_load_raw",
        bash_command="/opt/kh360-dataops/scripts/load_raw.sh",
    )
    freshness = BashOperator(
        task_id="dbt_source_freshness",
        bash_command=f"{DBT} source freshness --target prod",
    )
    run = BashOperator(
        task_id="dbt_run",
        bash_command=f"{DBT} run --target prod",
    )
    test = BashOperator(
        task_id="dbt_test",
        bash_command=f"{DBT} test --target prod",
    )
    publish = BashOperator(
        task_id="dbt_docs",
        bash_command=f"{DBT} docs generate --target prod",
    )

    el >> freshness >> run >> test >> publish

Vì chuỗi là run >> test >> publish, nếu dbt test fail thì publish không chạy và DAG báo đỏ — team nhận alert ngay. Đây là cách "circuit breaker" chất lượng vào orchestration đã bàn ở Airflow orchestration.

Bước 9 — CI trên Pull Request

Cuối cùng: không cho code hỏng vào main. Mỗi PR kích hoạt GitHub Actions chạy dbt build (= run + test gộp) trên một schema tạm tách biệt production. PR đỏ thì không merge được.

# .github/workflows/ci.yml  (minh hoạ — không đánh dấu chạy được)
name: dbt-ci
on:
  pull_request:
    branches: [main]
jobs:
  build:
    runs-on: ubuntu-latest
    env:
      DBT_HOST: ${{ secrets.DBT_HOST }}
      DBT_USER: ${{ secrets.DBT_CI_USER }}
      DBT_PASSWORD: ${{ secrets.DBT_CI_PASSWORD }}
      DBT_DEV_USER: ci_pr_${{ github.event.number }}   # schema riêng mỗi PR
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.11" }
      - run: pip install dbt-postgres
      - run: dbt deps
      - run: dbt build --target dev   # run + test cùng lúc

Mỗi PR build vào schema dev_ci_pr_<số> riêng nên các PR không đụng nhau; sau khi merge có thể dọn schema. Chi tiết chiến lược này ở CI/CD analytics.

Bước 10 — Chạy, kiểm tra, và xử lý khi test fail

Chạy toàn bộ lần đầu tại local:

dbt deps
dbt seed            # nếu có dữ liệu tĩnh
dbt run --select staging+     # build staging và mọi thứ downstream
dbt test
dbt docs generate

Đọc kết quả dbt test. Giả sử unique_dim_customer_customer_id FAIL với thông báo "Got 3 results". Quy trình xử lý:

1. Xem query lỗi: dbt lưu SQL biên dịch tại target/compiled/.../unique_dim_customer_customer_id.sql. Chạy tay để thấy customer_id nào bị trùng.
2. Chẩn đoán: thường do JOIN nở dòng (fan-out) — ví dụ một khách có nhiều account làm GROUP BY sai, hoặc source có id trùng.
3. Sửa đúng chỗ: nếu là fan-out, sửa logic aggregate trong dim_customer; nếu source bẩn, thêm bước dedup trong stg_customers.
4. Chạy lại: dbt build --select dim_customer cho tới khi xanh.
5. Không tắt test cho qua chuyện. Test đỏ đang bảo vệ báo cáo — bỏ test là bỏ tấm khiên.

Khi mọi test xanh ở local → commit → PR → CI xanh → merge → Airflow chạy đêm đầu tiên → mart sẵn sàng cho BI. Từ số 0, ta đã có một pipeline production đúng chuẩn: version hoá, có test, có lineage, có CI, có lịch chạy.

Use case thực tế

Bối cảnh: NCB cần báo cáo "Tổng giá trị giao dịch theo tháng, theo loại tiền" cho khối Bán lẻ, cập nhật mỗi sáng. Trước đây một analyst chạy tay 8 query trong SSMS mỗi thứ Hai, mất ~2 giờ, thỉnh thoảng lệch số vì sót một chi nhánh.

Áp dụng dự án này:

• Ngày 1–2: dựng repo, profiles.yml 2 target, khai báo 3 sources + freshness transactions error_after 24h.
• Ngày 3–5: viết stg_*, int_transactions_enriched, fct_transactions_monthly, dim_customer; thêm 9 generic tests + 1 singular test đối soát tổng.
• Ngày 6: đóng vào DAG kh360_daily (02:00), thêm CI trên PR.
• Ngày 7: chạy thử, singular test bắt được lệch 0,4 tỷ do một tài khoản có currency null lọt vào — sửa stg_accounts (loại/gán mặc định), test xanh.

Kết quả: thời gian ra báo cáo từ ~2 giờ/tuần thủ công xuống ~4 phút tự động mỗi đêm; lỗi lệch số về 0 nhờ test đối soát chặn trước; khi kiểm toán hỏi nguồn gốc, mở lineage graph trả lời trong 30 giây thay vì lục file. Quan trọng nhất: người mới đọc repo là hiểu toàn bộ pipeline, không còn "tri thức trong đầu anh Tuấn".

Ghi nhớ

• Mono-repo gom dbt + dags + tests + ci: một PR review được cả transform, orchestration và quality.
• Ba lớp model staging → intermediate → marts với quy ước tên stg_/int_/fct_/dim_ giúp repo tự-tài-liệu.
• Không hardcode secret: profiles.yml chỉ chứa tên env_var; tách target dev/prod, mỗi dev/PR một schema riêng.
• Khai báo sources + freshness để bắt lỗi nạp dữ liệu (EL) trước khi build ra số sai.
• ref()/source() tự dựng thứ tự build + lineage — không viết tay thứ tự chạy, docs sinh tự động.
• Kết hợp generic tests (not_null/unique/relationships/accepted_values) và singular test đối soát tổng — vũ khí số 1 trong ngân hàng.
• DAG theo chuỗi EL → run → test → publish; test fail thì DAG fail, publish không chạy.
• CI trên PR chạy dbt build vào schema tạm; PR đỏ không merge. Đây là chốt chặn cuối trước production.
• Khi test fail: đọc SQL biên dịch, chẩn đoán (thường fan-out JOIN), sửa đúng chỗ — tuyệt đối không tắt test cho qua.