dbt 3 — Sources, seeds & snapshots (SCD2)

Ở bài Tổng quan chúng ta đã thấy dbt lo phần chữ T trong ELT, và ở bài Models & materializations chúng ta đã biến những câu SELECT thành các bảng/view được vật chất hoá trong warehouse. Nhưng cả hai bài đó đều ngầm giả định một điều: dữ liệu thô đã nằm sẵn đâu đó trong warehouse rồi. Câu hỏi tự nhiên tiếp theo là: dbt biết dữ liệu thô nằm ở đâu bằng cách nào? Và làm sao lưu lại lịch sử thay đổi của nó?

Bài này trả lời trọn vẹn ba khối kiến thức nằm ở "biên" của một dbt project — nơi dbt tiếp xúc với thế giới bên ngoài:

• Sources — cách khai báo các bảng thô đã được nạp vào warehouse, và cách theo dõi độ tươi (freshness) của chúng.
• Seeds — cách đưa các file CSV nhỏ, tĩnh vào warehouse thành bảng tra cứu.
• Snapshots — cách bắt và lưu lịch sử thay đổi của bản ghi nguồn theo kiểu Slowly Changing Dimension Type 2 (SCD2).

Quy ước: sandbox của chúng ta là PostgreSQL và dbt không thực sự chạy ở đây. Toàn bộ khối yaml và sql bên dưới là minh hoạ để nắm cú pháp và tư duy, không phải để copy-paste chạy ngay. Nội dung bám theo dbt 1.x.

Bức tranh tổng: dữ liệu chảy từ đâu vào đâu

Trước khi đi vào chi tiết, hãy đặt ba khái niệm này vào đúng vị trí trong dòng chảy của một project. Sources và snapshots nằm sát nguồn; staging build trên chúng; rồi mới tới marts.

Điểm mấu chốt cần khắc vào đầu: snapshot chạy trước, run chạy sau. Thứ tự vận hành điển hình trong một lần refresh là dbt seed → dbt snapshot → dbt run. Lý do sẽ rõ khi ta hiểu snapshot làm gì: nó phải chụp lại trạng thái nguồn tại thời điểm hiện tại trước khi các model staging đọc và biến đổi dữ liệu.

Sources — khai báo điểm vào

Vấn đề mà sources giải quyết

Không có sources, model staging của bạn sẽ tham chiếu bảng thô bằng tên đầy đủ, cứng nhắc trong SQL:

-- Không dùng source (minh hoạ) — cứng và khó bảo trì
select * from raw.jaffle_shop.orders

Cách này có ba vấn đề: (1) nếu schema thô đổi tên (ví dụ từ raw sang raw_prod), bạn phải sửa ở hàng chục file; (2) dbt không biết bảng đó là "nguồn", nên lineage (đồ thị phụ thuộc) bị đứt ở đầu vào — bạn không thấy được model nào phụ thuộc bảng thô nào; (3) không có chỗ nào để gắn test hay kiểm tra chất lượng cho dữ liệu ngay tại điểm vào.

Sources sửa cả ba. Bạn khai báo bảng thô một lần trong một file .yml, rồi tham chiếu bằng hàm source().

Khai báo source trong .yml

# models/staging/jaffle_shop/_sources.yml  (minh hoạ)
version: 2

sources:
  - name: jaffle_shop          # tên logic của nhóm nguồn
    database: raw              # tuỳ chọn; mặc định = database của target
    schema: jaffle_shop        # tuỳ chọn; mặc định = name
    tables:
      - name: orders
      - name: customers
  - name: stripe
    tables:
      - name: payments

Vài quy tắc cần nhớ:

• name là bắt buộc — đây là định danh logic bạn sẽ gọi trong source().
• database là tuỳ chọn, mặc định lấy database của target đang chạy.
• schema là tuỳ chọn, nếu bỏ qua thì mặc định bằng name.
• tables là danh sách bảng, mỗi bảng ít nhất có name.

Tham chiếu bằng source()

Trong model staging, thay vì viết tên bảng cứng, bạn gọi:

-- models/staging/jaffle_shop/stg_orders.sql  (minh hoạ)
select
    id          as order_id,
    user_id     as customer_id,
    order_date,
    status
from {{ source('jaffle_shop', 'orders') }}

Khi biên dịch, {{ source('jaffle_shop', 'orders') }} sẽ được thay bằng raw.jaffle_shop.orders. Lợi ích thu về:

• Lineage tới nguồn: dbt hiểu stg_orders phụ thuộc source jaffle_shop.orders, nên đồ thị và trang docs sẽ vẽ đúng từ bảng thô đến marts.
• Đổi một nơi: đổi database/schema trong file .yml là mọi model tự cập nhật.
• Chỗ để gắn test và kiểm tra: bạn có thể gắn test (unique, not_null...) ngay trên cột của source — chủ đề chi tiết ở bài Tests & docs.

Source freshness — kiểm tra độ tươi

Đây là tính năng làm sources trở nên đắt giá trong vận hành thật. Dữ liệu thô do một pipeline nạp (EL) khác đổ vào warehouse — ví dụ Airbyte, Fivetran, hay một job Airflow. Nếu pipeline đó chết âm thầm, warehouse vẫn có bảng orders nhưng dữ liệu bên trong đã cũ vài ngày. dbt vẫn chạy ngon, chỉ có điều dashboard cuối cùng hiển thị số cũ mà không ai hay.

Source freshness là SLA nạp dữ liệu được mã hoá thành cấu hình. Bạn khai báo cột chứa dấu thời gian nạp (loaded_at_field) và ngưỡng cảnh báo:

# _sources.yml  (minh hoạ)
version: 2

sources:
  - name: jaffle_shop
    database: raw
    loaded_at_field: _etl_loaded_at        # cột thời điểm nạp
    freshness:
      warn_after:  {count: 12, period: hour}   # cũ hơn 12h → cảnh báo
      error_after: {count: 24, period: hour}   # cũ hơn 24h → lỗi
    tables:
      - name: orders
        # có thể override freshness cho riêng từng bảng
        freshness:
          warn_after:  {count: 6,  period: hour}
          error_after: {count: 12, period: hour}
      - name: customers
        freshness: null   # tắt kiểm tra freshness cho bảng này

Cấu trúc freshness gồm hai ngưỡng warn_after và error_after, mỗi ngưỡng là cặp count (số lượng) + period (đơn vị: hour, day...). freshness có thể khai báo ở cấp source rồi override ở cấp từng bảng; đặt freshness: null để tắt hẳn cho một bảng cụ thể.

Chạy kiểm tra bằng lệnh riêng:

dbt source freshness
dbt source freshness --select source:jaffle_shop
dbt source freshness --select source:jaffle_shop.orders

dbt tính khoảng cách từ giá trị loaded_at_field mới nhất tới hiện tại: vượt warn_after → cảnh báo (warn), vượt error_after → lỗi (error). Kết quả được ghi ra target/sources.json. Trong CI/CD, một lệnh dbt source freshness thất bại là tín hiệu sớm rằng "nguồn đang cũ" — bạn phát hiện sự cố nạp trước khi nó lan xuống báo cáo.

Seeds — dữ liệu tĩnh nhỏ dưới dạng CSV

Không phải dữ liệu nào cũng đến từ một pipeline EL. Có những bảng nhỏ, tĩnh, gần như không đổi mà bạn muốn quản lý cùng với code: bảng mã quốc gia, bảng phân loại/mapping, danh sách email test cần loại khỏi phân tích, danh sách ID nhân viên. Với những thứ này, dbt có seeds.

Seed là file CSV đặt trong thư mục seeds/ của project. Chạy dbt seed, dbt sẽ đọc CSV và tạo (hoặc nạp lại) một bảng trong warehouse. Vì file CSV nằm trong repo, nó được version-control và review y như code — đây chính là lý do seed hợp với dữ liệu tĩnh, thay đổi rất hiếm.

seeds/
  country_codes.csv

country_code,country_name
VN,Viet Nam
US,United States
JP,Japan

dbt seed                       # nạp tất cả seed
dbt seed --select country_codes
dbt seed --full-refresh        # nạp lại từ đầu khi đổi cấu trúc cột

Sau khi nạp, bảng tạo ra được tham chiếu bằng ref() giống hệt một model:

-- (minh hoạ)
select o.*, c.country_name
from {{ ref('stg_orders') }} o
left join {{ ref('country_codes') }} c using (country_code)

Bạn có thể cấu hình kiểu dữ liệu cột (column_types) hay ký tự phân tách (delimiter) trong dbt_project.yml, và có thể document/test seed trong file .yml như bảng thường.

Khi nào KHÔNG dùng seed — đây là chỗ nhiều người lạm dụng:

• Không dùng để nạp dữ liệu thô lớn xuất ra CSV — nạp CSV qua INSERT không hiệu năng với file lớn. Dữ liệu lớn là việc của pipeline EL + sources.
• Không đưa dữ liệu nhạy cảm (PII, mật khẩu) vào seed — nó nằm trong git.
• Không dùng cho dữ liệu thay đổi thường xuyên.

Nói ngắn gọn: seed là cho lookup/mapping tĩnh, nhỏ, không phải kho dữ liệu.

Snapshots — lưu lịch sử thay đổi (SCD Type 2)

Đây là phần đáng giá nhất của bài. Nó giải quyết một vấn đề mà sources và seeds không chạm tới: lịch sử.

Vì sao cần snapshot

Hệ thống nguồn thường ghi đè (mutate) tại chỗ. Bảng orders trong warehouse thô có cột status; khi đơn hàng chuyển từ placed sang shipped rồi completed, hệ nguồn cập nhật đè lên cùng một dòng. Lần nạp tiếp theo, giá trị cũ biến mất vĩnh viễn. Tương tự với "hạng khách hàng" (tier: bronze → silver → gold) — nguồn chỉ giữ trạng thái hiện tại.

Nhưng phân tích lại thường cần biết quá khứ: "Tại thời điểm đơn hàng này được đặt, khách đang ở hạng nào?", "Một đơn hàng trung bình mất bao lâu để chuyển từ placed sang shipped?". Nếu nguồn đã ghi đè, những câu hỏi này không thể trả lời được nữa. Bạn cần chủ động chụp lại trạng thái theo thời gian — đó chính là Slowly Changing Dimension Type 2 (SCD2): mỗi khi một thuộc tính đổi, thay vì ghi đè, ta thêm một phiên bản mới của dòng và đóng khoảng hiệu lực của phiên bản cũ. (Khái niệm SCD ở tầng warehouse modeling / BI đã được bàn tổng quát; ở đây dbt hiện thực hoá nó tự động.)

snapshot làm gì cụ thể

Bạn khai báo một snapshot chỉ về nguồn bạn muốn theo dõi lịch sử. Mỗi lần chạy dbt snapshot:

• Lần đầu: dbt tạo bảng snapshot chứa toàn bộ dòng hiện có, gắn dbt_valid_from = thời điểm hiện tại, dbt_valid_to = NULL (nghĩa là "còn hiệu lực").
• Các lần sau: dbt so sánh nguồn với ảnh chụp gần nhất. Với dòng đã thay đổi, nó đóng phiên bản cũ (điền dbt_valid_to = thời điểm hiện tại) và chèn một dòng mới với dbt_valid_from mới, dbt_valid_to = NULL. Dòng mới hoàn toàn thì được chèn như bản ghi mới.

Các cột meta dbt tự thêm vào bảng snapshot:

Vòng đời một dòng trong snapshot

Kết quả: bảng snapshot giữ đầy đủ chuỗi trạng thái theo thời gian, mỗi dòng là một "lát cắt" hiệu lực. Muốn lấy trạng thái hiện tại, lọc dbt_valid_to is null; muốn biết trạng thái tại một thời điểm T, lọc dbt_valid_from <= T and (dbt_valid_to is null or dbt_valid_to > T).

Hai chiến lược: timestamp vs check

dbt cần một cách để phát hiện "dòng có đổi không". Có hai chiến lược:

Chiến lược timestamp (khuyến nghị). Dựa trên một cột dấu thời gian cập nhật của nguồn (thường là updated_at). Nếu updated_at của dòng nguồn mới hơn updated_at đã lưu, dbt coi là đã đổi và tạo phiên bản mới. Ưu điểm: chỉ theo dõi một cột, tự chịu được việc nguồn thêm/bớt cột. Điều kiện: nguồn phải có một cột updated_at đáng tin — luôn được cập nhật khi dòng đổi.

Chiến lược check. Dùng khi nguồn không có cột updated_at tin cậy. dbt so sánh giá trị của một tập cột bạn chỉ định (check_cols) giữa lần chạy này và ảnh chụp trước; khác nhau ở bất kỳ cột nào trong danh sách thì tạo phiên bản mới. Bạn có thể liệt kê cột cụ thể hoặc dùng check_cols: 'all' để so mọi cột.

Khai báo snapshot (block Jinja)

Trong dbt 1.x, cách phổ biến là viết snapshot trong file .sql đặt ở thư mục snapshots/, dùng block {% snapshot %}:

-- snapshots/orders_snapshot.sql  (minh hoạ) — chiến lược timestamp
{% snapshot orders_snapshot %}

  {{
    config(
      target_schema='snapshots',
      unique_key='order_id',
      strategy='timestamp',
      updated_at='updated_at'
    )
  }}

  select * from {{ source('jaffle_shop', 'orders') }}

{% endsnapshot %}

Và bản dùng chiến lược check khi không có updated_at đáng tin:

-- snapshots/customers_snapshot.sql  (minh hoạ) — chiến lược check
{% snapshot customers_snapshot %}

  {{
    config(
      target_schema='snapshots',
      unique_key='customer_id',
      strategy='check',
      check_cols=['tier', 'status']
    )
  }}

  select * from {{ source('jaffle_shop', 'customers') }}

{% endsnapshot %}

Các config bắt buộc/quan trọng:

• unique_key — khoá phải thực sự duy nhất trên nguồn; dbt dùng nó để "khớp" dòng qua các lần chụp. Nên gắn thêm test unique cho nguồn để phòng ngừa.
• strategy — timestamp hoặc check.
• updated_at — bắt buộc với timestamp; là cột dấu thời gian cập nhật.
• check_cols — bắt buộc với check; danh sách cột hoặc 'all'.
• target_schema — schema đặt bảng snapshot (thường tách riêng, ví dụ snapshots).

Chạy và tham chiếu

dbt snapshot
dbt snapshot --select orders_snapshot

Snapshot được chạy định kỳ (thường hàng giờ đến hàng ngày) — nó là dạng CDC theo lô: mỗi lần chạy chụp một trạng thái. Nếu bạn thấy mình cần chụp rất dày, đó là dấu hiệu nên cân nhắc một giải pháp CDC thật sự thay vì snapshot. Sau khi chạy, bảng snapshot được tham chiếu xuống dưới bằng ref() như một model bình thường:

-- (minh hoạ) staging đọc từ snapshot để lấy trạng thái hiện hành
select * from {{ ref('orders_snapshot') }}
where dbt_valid_to is null

Use case thực tế: giữ lịch sử hạng khách hàng

Hãy dựng một tình huống cụ thể. Hệ CRM lưu bảng customers với cột tier (bronze/silver/gold). CRM ghi đè tier mỗi khi khách lên/xuống hạng — không giữ lịch sử. Đội phân tích cần trả lời: "Doanh thu tháng 3 đến từ khách ở hạng nào tại thời điểm mua?" Nếu chỉ có trạng thái hiện tại, một khách nay là gold sẽ khiến mọi đơn quá khứ của họ bị tính nhầm vào gold, kể cả khi lúc mua họ mới là bronze.

Giải pháp bằng dbt:

1. Khai báo customers là source (jaffle_shop.customers), gắn freshness để biết CRM có nạp đều không.
2. Tạo snapshot customers_snapshot trên source đó. Nếu CRM có updated_at tin cậy, dùng strategy='timestamp'; nếu không, dùng strategy='check' với check_cols=['tier'].
3. Lên lịch dbt snapshot chạy hằng ngày (đứng trước dbt run). Mỗi lần khách đổi hạng, snapshot đóng phiên bản cũ và mở phiên bản mới với dbt_valid_from/dbt_valid_to.
4. Ở marts, khi tính doanh thu, join đơn hàng với snapshot theo khoảng hiệu lực: ghép order_date vào khoảng [dbt_valid_from, dbt_valid_to) để lấy đúng hạng tại thời điểm mua.

-- marts/fct_orders_with_tier.sql  (minh hoạ)
select
    o.order_id,
    o.customer_id,
    o.order_date,
    o.amount,
    c.tier as tier_at_order_time
from {{ ref('stg_orders') }} o
left join {{ ref('customers_snapshot') }} c
    on o.customer_id = c.customer_id
   and o.order_date >= c.dbt_valid_from
   and (o.order_date < c.dbt_valid_to or c.dbt_valid_to is null)

Từ chỗ không thể trả lời câu hỏi lịch sử, giờ ta trả lời chính xác — chỉ nhờ đặt một snapshot đúng chỗ và chạy nó đều đặn. Đây là lý do snapshot được xem như "bảo hiểm lịch sử": bạn không lấy lại được quá khứ đã bị nguồn ghi đè, nên phải bắt đầu chụp càng sớm càng tốt.

Vị trí trong project & thứ tự chạy

Ghép lại toàn bộ, một dbt project điển hình sắp xếp như sau:

• Sources khai báo trong .yml cạnh các model staging tương ứng.
• Seeds nằm trong seeds/ (CSV tĩnh, nhỏ).
• Snapshots nằm trong snapshots/, đặt gần nguồn (chụp thẳng từ source()), không trộn chung thư mục với models.
• Staging build trên sources/snapshots — mỗi source table thường ứng với một staging model.
• Marts build trên staging.

Thứ tự vận hành một lần refresh: dbt seed → dbt snapshot → dbt run (và xen dbt source freshness để canh SLA, dbt test để kiểm chất lượng). Snapshot phải chạy trước run, vì run/staging đọc dữ liệu — nếu snapshot chạy sau, bạn đã bỏ lỡ việc chụp trạng thái tại thời điểm run.

Best practice

• Một staging model cho một source table. Giữ tầng staging "mỏng" — chỉ đổi tên cột, ép kiểu, làm sạch nhẹ — và ánh xạ 1–1 với bảng nguồn.
• Đặt snapshot gần nguồn, chụp trực tiếp từ source() chứ không từ model đã biến đổi, để lịch sử phản ánh đúng dữ liệu thô.
• unique_key phải thực sự duy nhất — sai điểm này là snapshot khớp nhầm dòng và lịch sử hỏng. Gắn test unique/not_null cho khoá.
• Ưu tiên strategy='timestamp' khi nguồn có updated_at tin cậy; chỉ dùng check khi buộc phải.
• Bắt đầu snapshot sớm — bạn không chụp lại được quá khứ đã mất.
• Dùng seed đúng mục đích — mapping/lookup tĩnh nhỏ, không phải kho dữ liệu, không PII.
• Bật source freshness cho các nguồn quan trọng để bắt lỗi nạp trước khi nó lan xuống báo cáo.

Ghi nhớ

• Sources = khai báo bảng thô một lần trong .yml, gọi bằng source('nhóm','bảng'). Đổi được lineage, đổi một nơi, có chỗ gắn test.
• Freshness = SLA nạp: loaded_at_field + warn_after/error_after, kiểm bằng dbt source freshness.
• Seeds = CSV tĩnh nhỏ trong seeds/, nạp bằng dbt seed, gọi bằng ref(). Cho lookup, không cho dữ liệu lớn.
• Snapshots = SCD2 tự động: dbt thêm dbt_valid_from/dbt_valid_to; chiến lược timestamp (dựa updated_at) vs check (so cột); chạy dbt snapshot định kỳ, trước dbt run.
• Thứ tự: seed → snapshot → run. Snapshot & sources ở sát nguồn; staging build trên chúng; marts trên staging.

Bài tiếp theo — Tests & docs — sẽ chỉ cách gắn kiểm thử chất lượng lên chính những sources, seeds và snapshots này, và sinh tài liệu + đồ thị lineage tự động cho toàn bộ project.