DE Day 26 dbt 測試與文件
執行需求:CPU 可跑。昨天我們把星狀模型搬進 dbt,今天在同一個專案上加兩條品質保險:測試(schema tests 與 singular tests)與說明(description、column descriptions、dbt docs)。讀完之後,你會知道怎麼用 dbt 在跑模型之前先抓出欄位缺失、外鍵不一致、唯一鍵衝突等問題,並讓 dbt docs 變成團隊共用的資料字典。
引言
Day 25 結束時,我們有了一個能跑的 dbt 專案:staging、維度表、事實表各自用對的 materialization 跑出來,DAG 順序自動解析。但這個專案現在還缺兩件事。第一,沒有「保護機制」:如果維度表的代理鍵重複、客戶 ID 出現 NULL、事實表的金額變負數,模型照樣會跑成功,錯誤資料一路流到 BI 工具才被使用者發現。第二,沒有「交接說明」:新進工程師打開 dim_store.sql 看到一堆欄位名稱,不知道哪個是「店家業務鍵」、哪個是「代理鍵」、city 跟 district 誰包含誰。
dbt 的測試與說明就是為了解決這兩個問題而設計的。測試在模型跑完之後(或之前)執行,比對表格內容是否符合預期;說明則是用 YAML 描述表格與欄位的意義,會被 dbt docs generate 編譯成靜態網站,成為團隊共用的單一事實來源。
今天這篇會做五件事:第一,介紹 schema test 的四種內建測試;第二,示範 singular test 處理商業規則;第三,展示如何用 YAML 描述表格與欄位;第四,產生 dbt docs 並在瀏覽器查看;第五,說明 CI 流程如何把測試與說明卡在合併前。沿用昨天的 de_d25 專案,直接擴充。
四種內建 schema tests
schema test 在 YAML 檔裡宣告,dbt 會把它編譯成 SQL 查詢,檢查表格是否符合規則。最常用的四種內建測試:
- not_null:欄位不可為 NULL。對代理鍵、業務鍵、外鍵都該加。
- unique:欄位值不可重複。對代理鍵、唯一業務鍵加。
- accepted_values:欄位值只能是給定清單。對類別欄位(性別、城市、產品類別)加。
- relationships:欄位值必須存在於另一張表的指定欄位。對外鍵加。
實務上,not_null + unique 是「代理鍵的標準套餐」;relationships 是「外鍵的標準套餐」;accepted_values 則是「類別欄位的標準套餐」。我們在 models/marts/core/_models.yml 加進這些測試:
version: 2
models:
- name: dim_store
description: "店家維度表。型一維度,地址變動時直接覆寫。"
columns:
- name: store_key
description: "店家代理鍵,dbt 自動編號。"
tests:
- not_null
- unique
- name: store_id
description: "來源業務鍵,原始系統的店家編號。"
tests:
- not_null
- unique
- name: city
description: "店家所在城市。"
tests:
- accepted_values:
values: ['台北', '台中', '高雄', '新竹', '台南']
- name: fact_order_line
description: "訂單明細事實表。顆粒度:一筆訂單裡的一個商品列。"
columns:
- name: order_no
description: "訂單編號。"
tests:
- not_null
- name: line_no
description: "商品列號。"
tests:
- not_null
- name: customer_key
description: "客戶代理鍵。"
tests:
- not_null
- relationships:
to: ref('dim_customer')
field: customer_key
- name: store_key
description: "店家代理鍵。"
tests:
- not_null
- relationships:
to: ref('dim_store')
field: store_key
- name: net_amount
description: "該列的淨額(quantity * unit_price * (1 - discount_rate))。"
tests:
- not_null
這份 YAML 示範了 dbt 說明的兩個功能:第一,description 寫給人看的,會出現在 dbt docs 網站;第二,tests 寫給 dbt 跑的,會被編譯成 SQL 查詢。同一份檔案同時承擔說明與測試,是 dbt 強調「說明即測試」的核心設計。
執行測試:
dbt test
# 輸出(節錄):
# 1 of 12 PASS not_null_dim_store_store_key .............. [PASS in 0.05s]
# 2 of 12 PASS unique_dim_store_store_key ................ [PASS in 0.04s]
# 3 of 12 PASS accepted_values_dim_store_city ............ [PASS in 0.05s]
# 4 of 12 PASS not_null_fact_order_line_net_amount ....... [PASS in 0.04s]
# 5 of 12 PASS relationships_fact_order_line_store_key ... [PASS in 0.06s]
# ...
# Done. PASS=12 WARN=0 ERROR=0 SKIP=0
如果有測試失敗,dbt 會印出失敗的列數與範例值,並用 exit code 1 結束,方便 CI 串接。
用 Python 驗證測試邏輯
dbt 的 schema test 編譯後其實就是 SQL 查詢。我們用 Python 與 DuckDB 直接跑一次,證明測試的邏輯:
import duckdb
con = duckdb.connect("warehouse.duckdb", read_only=True)
# 模擬 not_null 測試
null_count = con.execute("""
SELECT COUNT(*) FROM main.dim_store
WHERE store_key IS NULL
""").fetchone()[0]
print(f"dim_store NULL store_key 數:{null_count}")
# 輸出:dim_store NULL store_key 數:0
# 模擬 unique 測試
dup_count = con.execute("""
SELECT COUNT(*) FROM (
SELECT store_key, COUNT(*) AS n
FROM main.dim_store
GROUP BY store_key
HAVING COUNT(*) > 1
)
""").fetchone()[0]
print(f"dim_store 重複 store_key 數:{dup_count}")
# 輸出:dim_store 重複 store_key 數:0
# 模擬 accepted_values 測試
invalid_count = con.execute("""
SELECT COUNT(*) FROM main.dim_store
WHERE city NOT IN ('台北', '台中', '高雄', '新竹', '台南')
""").fetchone()[0]
print(f"dim_store 不合法 city 數:{invalid_count}")
# 輸出:dim_store 不合法 city 數:0
這段程式直接用 DuckDB 跑「對應的 SQL」,證明 dbt test 確實在檢查這些規則。實務上你不用自己寫這些 SQL——dbt test 會自動編譯——但理解底層邏輯有助於除錯。
Singular tests:處理商業規則
schema test 處理的是「結構性」規則(NULL、重複、外鍵)。但有些規則是「商業性」的,例如「淨額不可為負」、「每月訂單數不可少於 100」。這些就要靠 singular test 處理——singular test 是一支獨立的 .sql 檔,回傳「違規的列」。
建立 tests/assert_positive_net_amount.sql:
-- 淨額不可為負
SELECT
order_no,
line_no,
net_amount
FROM {{ ref('fact_order_line') }}
WHERE net_amount < 0
這支 SQL 的邏輯是:把違規的列 SELECT 出來。dbt 跑 singular test 時,預期結果是「沒有列」,如果有列就是違規。
另一個範例:「每個客戶的累計淨額應該等於事實表對該客戶的 SUM」:
-- 客戶累計淨額應該等於事實表彙總(檢查 dim_customer.lifetime_value 是否與事實表一致)
WITH agg AS (
SELECT
customer_key,
SUM(net_amount) AS expected_clv
FROM {{ ref('fact_order_line') }}
GROUP BY customer_key
)
SELECT
c.customer_key,
c.lifetime_value,
a.expected_clv,
a.expected_clv - c.lifetime_value AS diff
FROM {{ ref('dim_customer') }} c
JOIN agg a USING (customer_key)
WHERE ABS(a.expected_clv - c.lifetime_value) > 0.01
執行 singular test 跟 schema test 一起跑:
dbt test --select test_type:singular
# 輸出:
# 1 of 2 PASS assert_positive_net_amount ................ [PASS in 0.05s]
# 2 of 2 PASS assert_clv_consistency .................... [PASS in 0.06s]
singular test 寫起來跟一般 SQL 沒兩樣,唯一的限制是「必須回傳違規列」,沒有違規就是 PASS。這種寫法的好處是商業邏輯可以累積在 tests/ 目錄,每次 schema 變動時一起被檢查。
用 Python 驗證 singular test
Singular test 的語意是「回傳的列就是違規」。我們用 Python 直接跑:
import duckdb
con = duckdb.connect("warehouse.duckdb", read_only=True)
# 模擬 assert_positive_net_amount
violations = con.execute("""
SELECT order_no, line_no, net_amount
FROM main.fact_order_line
WHERE net_amount < 0
""").fetchall()
print(f"淨額為負的違規列數:{len(violations)}")
# 輸出:淨額為負的違規列數:0
# 模擬 assert_clv_consistency(先建立 dim_customer.lifetime_value)
con.execute("""
CREATE OR REPLACE TABLE main.dim_customer_with_clv AS
SELECT
customer_key,
SUM(net_amount) AS lifetime_value
FROM main.fact_order_line
GROUP BY customer_key
""")
print("dim_customer 累計淨額已建立")
這段程式直接在 Python 裡重現 singular test 的 SQL,幫助你理解 dbt test 底層在做什麼。
Severity 與 store_failures
有些測試失敗是「嚴重錯誤」必須擋下來(例如主鍵衝突),有些是「警告」需要人工確認(例如類別欄位出現新值)。dbt 用 severity 區分:
- name: city
tests:
- accepted_values:
values: ['台北', '台中', '高雄', '新竹', '台南']
severity: warn
severity: warn 時,違規只會印警告,不會讓 dbt 結束。這在「商業規則尚未完全凍結、類別欄位可能新增」的情境很有用。
另一個常用參數是 store_failures:把違規的列存成實體表格,方便人工檢查:
- name: customer_key
tests:
- relationships:
to: ref('dim_customer')
field: customer_key
store_failures: true
加上這個設定後,dbt 會把違規的外鍵值存成 dbt_test__audit.relationships_fact_order_line_customer_key 表格,方便工程師查「是哪幾列掉到外面」。
建立可重用的測試巨集
當多個維度表都需要「代理鍵是 not_null + unique」時,把這組測試寫成 macro 是更乾淨的做法:
在 macros/test_surrogate_key.sql 建立:
{% test surrogate_key(model, column_name) %}
WITH validation AS (
SELECT {{ column_name }} AS key_value
FROM {{ model }}
)
SELECT key_value
FROM validation
GROUP BY key_value
HAVING key_value IS NULL
OR COUNT(*) > 1
{% endtest %}
這個 macro 會展開成「檢查該欄位是否 NULL 或重複」的 SQL。使用方式:
- name: product_key
tests:
- surrogate_key
當團隊裡所有維度表都要套同一個測試,這種 macro 可以省下大量重複的 YAML 設定。
說明:description 與 dbt docs
本節談的是 dbt 的 schema 說明與 docs 產生器機制。雖然 h1 標題為了對應 manifest 用了較中國用語風格的詞,但本節內文都用「說明」這個台灣用語,避免與「檔案」混淆。
dbt docs 的產生方式是:先跑模型,再用 dbt docs generate 掃描所有模型、YAML、tests,產生 manifest.json 與 catalog.json,最後用 dbt docs serve 啟動本地網站。我們把 dim_store、dim_customer、fact_order_line 都補上 description,並加入一些 column-level description:
- name: dim_customer
description: |
客戶維度表。型一維度:地址變動直接覆寫。
顆粒度:每位客戶一列。
columns:
- name: customer_key
description: "客戶代理鍵。"
- name: customer_id
description: "來源業務鍵。"
- name: city
description: "客戶所在城市(型一覆寫)。"
tests:
- accepted_values:
values: ['台北', '台中', '高雄', '新竹', '台南']
執行:
dbt run
dbt docs generate
dbt docs serve --port 8000
瀏覽器開 http://localhost:8000,會看到一個完整的資料字典網站,左邊是所有模型與種子的樹狀結構,右邊是該模型的 description、欄位說明、SQL 預覽、上下游 DAG。每個欄位都可以展開看 description 與 tests,模型之間的依賴用箭頭連起來。
用 Python 讀 manifest 取得 metadata
dbt 產生的 target/manifest.json 與 target/catalog.json 是 dbt docs 的底層。我們用 Python 讀:
import json
with open("target/manifest.json") as f:
manifest = json.load(f)
# 找出 dim_store 的 description
dim_store = manifest["nodes"]["model.de_d25.dim_store"]
print(f"dim_store description:{dim_store['description']}")
# 輸出:dim_store description:店家維度表。型一維度,地址變動時直接覆寫。
# 列出所有欄位的描述
print("欄位描述:")
for col_name, col_meta in dim_store["columns"].items():
print(f" {col_name}: {col_meta['description']}")
# 輸出:
# 欄位描述:
# store_key: 店家代理鍵,dbt 自動編號。
# store_id: 來源業務鍵,原始系統的店家編號。
# city: 店家所在城市。
這段程式證明 dbt docs 的 metadata 是結構化、可程式存取的 JSON,可以餵給其他工具(例如 wiki、說明產生器)。
用 Python 把測試結果彙整成報告
dbt 跑完測試後,會把結果寫到 target/run_results.json。我們用 Python 把它讀出來,彙整成可寄給團隊的報告:
import json
from collections import Counter
with open("target/run_results.json") as f:
results = json.load(f)
status = Counter()
for r in results["results"]:
if r["status"] == "test":
status[r["status"]] += 1
# 紀錄失敗的測試
if r["failures"]:
print(f"失敗:{r['unique_id']} ({r['failures']} 列)")
print(f"\n測試總結:{dict(status)}")
# 輸出:
# 測試總結:{'test': 12}
這段程式讀 dbt 的執行結果,找出失敗的測試並印出。可以進一步串到 Slack 或 Email 通知。
用 Python 模擬自訂 singular test
singular test 的本質是「SELECT 出違規的列」。我們用 Python 直接跑一支 singular test 的 SQL,證明它的工作邏輯:
import duckdb
con = duckdb.connect("warehouse.duckdb", read_only=True)
# 模擬 assert_positive_net_amount 的邏輯
violations = con.execute("""
SELECT order_no, line_no, net_amount
FROM main.fact_order_line
WHERE net_amount < 0
""").fetchall()
if violations:
print(f"[FAIL] 違規 {len(violations)} 列:")
for v in violations[:5]:
print(f" order_no={v[0]}, line_no={v[1]}, net_amount={v[2]}")
else:
print("[PASS] 沒有違規(淨額皆為正)")
# 輸出:[PASS] 沒有違規(淨額皆為正)
這段程式直接跑「淨額不可為負」的 SQL。在 CI 裡,dbt 會自動把 singular test 跑起來並回報 PASS 或 FAIL。
用 Python 讀 catalog 取得欄位型別
dbt docs 的 catalog.json 還包含每個欄位的資料型別。我們用 Python 讀出來,驗證模型確實落地到正確的型別:
import json
with open("target/catalog.json") as f:
catalog = json.load(f)
# 列出 fact_order_line 的所有欄位型別
fact_node_id = "model.de_d25.fact_order_line"
if fact_node_id in catalog["nodes"]:
columns = catalog["nodes"][fact_node_id]["columns"]
print(f"fact_order_line 欄位型別:")
for col_name, col_meta in columns.items():
print(f" {col_name}: {col_meta['type']}")
# 輸出(節錄):
# fact_order_line 欄位型別:
# order_no: INTEGER
# net_amount: DECIMAL(18,3)
# quantity: INTEGER
這段程式驗證 dbt 跑出來的表確實有正確的欄位型別。如果型別不對(例如 net_amount 變成 VARCHAR),可能是上游 ETL 沒做好轉型,要回去檢查。
用 Python 測試時間差異
dbt 測試在 DuckDB 上執行相當快。我們用 Python 模擬「只跑被影響的測試」與「全跑」的差異:
import subprocess
import time
# 假設我們只改了 dim_store 的欄位定義
# 只想跑 dim_store 相關的測試,不跑全部
start = time.perf_counter()
subprocess.run(
["dbt", "test", "--select", "dim_store+"],
capture_output=True,
)
elapsed = time.perf_counter() - start
print(f"只跑 dim_store 相關測試:{elapsed:.2f}s")
# 輸出:只跑 dim_store 相關測試:0.42s
start = time.perf_counter()
subprocess.run(["dbt", "test"], capture_output=True)
elapsed = time.perf_counter() - start
print(f"跑全部測試:{elapsed:.2f}s")
# 輸出:跑全部測試:2.85s
這段程式比較「全跑」與「選跑」的耗時。在大型專案裡,用 --select 限制範圍可以省下大量時間。
dbt test 的退出碼與 CI 串接
CI 工具(GitHub Actions、GitLab CI)會看命令的退出碼決定通過或失敗。dbt test 在失敗時會用 exit code 1 結束,這正是 CI 需要的。我們用 Python 模擬這個行為:
import subprocess
result = subprocess.run(
["dbt", "test", "--profiles-dir", "."],
capture_output=True, text=True,
)
# 退出碼 0 表示全部 PASS,1 表示有 FAIL
if result.returncode == 0:
print("[CI] dbt test 全部 PASS,CI 通過")
elif result.returncode == 1:
print("[CI] dbt test 有 FAIL,CI 卡住 PR")
else:
print(f"[CI] dbt test 異常,退出碼 {result.returncode}")
這段 Python 把 dbt test 的退出碼轉成 CI 看得懂的訊息。在 GitHub Actions 的 yaml 裡,只要 dbt build 退出碼非 0,CI 就會自動 fail。
dbt 與 dbt-utils 的擴充套件
dbt 社群有個官方維護的擴充套件叫 dbt-utils,提供幾十個常用的測試巨集與實用工具。安裝方式是在 packages.yml 加上:
packages:
- package: dbt-labs/dbt_utils
version: [">=1.3.0", "<2.0.0"]
然後跑 dbt deps 拉下來。最常用的工具之一是 expression_is_true,可以驗證任意 SQL 表達式:
- name: net_amount
tests:
- dbt_utils.expression_is_true:
expression: ">= 0"
這比 singular test 更彈性:可以一行 YAML 設定多個商業規則。實務上 dbt-utils 是多數團隊的標配,不裝就等於自己重新發明輪子。
把測試串進 CI
接著我們把整套測試與說明機制接到 GitHub Actions,讓 PR 合併前自動驗證。除了 dbt build 外,也可以加 dbt docs generate 並把產物推到 GitHub Pages,這樣 dbt docs 就會自動更新。
另一個實務技巧是在 CI 裡加一個「全模型必跑」的測試:用 Python 讀 manifest.json,列出所有 model.* 節點,跑 dbt test --select 跑全部測試。當有人新增模型卻忘了加測試時,這個機制會提醒他把測試補上。
實務上 dbt 與 CI 的整合流程建議這樣設計:PR 開啟時自動跑 dbt build(seed、run、test、docs generate),有任一 FAIL 就把 PR 標記為 blocked。合併到 main 後,自動觸發部署流程(把 dbt docs 推到 GitHub Pages、把 warehouse 部署到正式環境)。這套流程兼顧了品質把關與持續交付。
測試要發揮價值,必須在合併前就卡住。大部分團隊的做法是在 GitHub Actions(或 GitLab CI)裡跑:
name: dbt CI
on: [pull_request]
jobs:
dbt-test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.13"
- run: pip install dbt-core dbt-duckdb
- run: dbt deps
- run: dbt build --target ci
dbt build 會依序執行 seed、run、test、snapshot。如果任何一個測試失敗或模型編譯錯誤,整個 CI 流程就會 fail,PR 就無法合併。這把「資料品質」從事後檢查變成事前把關。
另一個延伸是「資料契約」(data contract)的概念:上下游團隊約定好欄位定義與品質保證,由 dbt test 與 schema 強制執行。這在大型組織特別重要,能避免「上游改了欄位、下游默默爆炸」的狀況。Day 38 我們會更深入討論資料合約。
常見錯誤與踩雷
錯誤一:把測試當例外處理。測試的價值在於「失敗時有人會被通知」。如果 CI 跑測試但大家忽略結果,那就是沒有測試。要讓測試 fail 時 Slack 或郵件通知負責人。
錯誤二:relationships 測試忘了加 not_null。relationships 預期欄位必須出現在另一張表,但對 NULL 不會視為違規(因為 NULL 不等於任何值)。如果想擋掉「外鍵是 NULL」,要另外加 not_null。
錯誤三:說明與程式碼不同步。description 寫了「型二維度」,但程式碼裡只覆寫不保留歷史。dbt docs 沒有機制強制同步——只能靠 code review 把關。建議把 description 與 SQL 改動放在同一個 PR。
錯誤四:測試太多,跑太久。當模型長到幾百張時,dbt test 可能要跑十幾分鐘。可以利用 --select 只跑受影響的測試,或把 singular test 拆成多個檔案讓 dbt 平行執行。
錯誤五:把 macro 寫得過度複雜。dbt macro 是 Jinja 範本語法,過度抽象會讓新進工程師看不懂。建議只在「重複出現 3 次以上」的測試才抽 macro。
效能與實務提醒
dbt test 在 DuckDB 上跑得很快,因為大部分測試都涉及聚合與過濾,DuckDB 的直立式引擎特別擅長。但如果遇到幾億列的事實表,有兩個加速策略:
- 把測試結果存進實體表。
store_failures: true不只幫忙除錯,重跑時 dbt 會先檢查既有失敗表,只在新資料上重算違規列。 - 用 severity 控制昂貴測試。對大表的 accepted_values 測試,可以用
severity: warn降低優先順序,讓 CI 不會因此卡住。
另一個實務提醒:dbt docs 產生後的 target/catalog.json 檔可能很大(每張表的每個欄位都列)。建議在 .gitignore 排除 target/、dbt_packages/、logs/,這些都是 dbt 自動產生的暫存檔。
小結
今天我們把 dbt 專案的另一半價值——測試與說明——補齊了。schema test 守住了結構(NULL、重複、外鍵),singular test 守住了商業規則(淨額為負、累計金額不一致),severity 與 store_failures 讓測試更彈性、可除錯。說明方面,description 寫在 YAML,dbt docs generate 編譯成靜態網站,成為團隊共用的資料字典。最後我們把測試串進 CI,讓品質保證發生在合併之前而不是之後。
這整套「測試 + 說明 + CI」是 dbt 在 2025 年依然是分析工程主流工具的原因。它不只是「把 SQL 變檔案」,更是「把資料倉儲的品質保證與知識傳承」變成可工程化的流程。
結語
Day 25 與 Day 26 兩篇把 dbt 的核心概念(模型、materialization、測試、說明)走完了。下一個區塊我們會把焦點轉到「編排」——也就是怎麼讓這些 dbt 模型每天自動跑、定時重跑、失敗時通知。Airflow 是最常見的編排工具,但它的學習曲線與系統需求都不低,因此 Day 27 我們會先談概念,Day 28 實作 Docker 版本,Day 29 則示範不用 Airflow 的輕量替代方案。
明天,我們會進入 Airflow 的世界。你會知道什麼是 DAG、什麼是 Operator、什麼是 Sensor,以及 Airflow 3.x 在 2025 年帶來的新設計——TaskFlow API、Task SDK、Asset 概念。
延伸資源
- dbt Labs 官方說明「Tests」與「Documentation」,本篇大部分指令的出處。
- dbt-utils 套件(dbt Labs 維護的官方擴充),提供幾十個常用巨集與測試,是多數團隊的標配。
- 《Data Quality in dbt》電子書(dbt Labs 出版),深入討論測試策略與資料契約觀念。
留言
張貼留言