dbt資料轉換實戰2026：5種建模模式建構現代資料棧轉換層
AI与大数据
SQL腳本散落、轉換邏輯混亂、資料品質無人把關
凌晨3點，營運總監質問為何昨日GMV報表資料對不上——上游訂單表改了欄位名，20個SQL預存程序沒人更新，髒資料一路灌入BI看板；資料工程師寫了300行巢狀子查詢的「義大利麵SQL」，新人接手完全看不懂；ETL腳本散落在15個Cron Job裡，上游表結構變更下游全掛，但沒人知道。2026年，**dbt Core 1.8+**帶來了Python模型支援、改進的增量策略和更強大的單元測試——從專案搭建到生產部署，一套體系全搞定。
本文將從5種資料建模模式出發，帶你完成dbt專案搭建與模型設計→增量模型與快照→資料測試與品質門禁→巨集與自訂物化→dbt Cloud CI/CD與生產部署的全鏈路實戰，每一步都有完整可執行的SQL和Python程式碼。
dbt核心概念
| 概念 | 說明 |
|---|---|
| Model | dbt的核心抽象，一個.sql檔案定義一次資料轉換邏輯，編譯後執行 |
| Materialization | 模型的物化策略：table（全量重建）、view（虛擬檢視）、incremental（增量追加）、ephemeral（內聯CTE） |
| Incremental Model | 增量模型，只處理新增或變更資料，避免全量重新整理，是大資料量場景的核心策略 |
| Snapshot | 快照，追蹤維度表的歷史變更（SCD Type 2），自動記錄生效和失效時間 |
| Test | 資料測試，斷言資料品質規則（唯一性、非空、參照完整性、自訂範圍） |
| Macro | Jinja巨集，可復用的SQL程式碼片段，類似函式，支援參數化和條件邏輯 |
| Seed | CSV種子檔案，將靜態資料載入到資料倉儲，適合小維度表和對應關係 |
| Source | 宣告上游資料來源，配合freshness偵測資料新鮮度，建構資料血緣起點 |
| Ref | 模型間參照函式ref('model_name')，dbt自動解析依賴並建構DAG |
| Documentation | dbt文件產生，在YAML和MD檔案中描述模型、欄、資料來源，自動產生資料字典 |
| Packages | dbt套件，復用社群或團隊的巨集、模型和測試，類似Python的pip套件 |
| dbt Cloud | dbt官方託管平台，提供CI/CD、排程、文件託管和企業級協作 |
問題分析：資料轉換的5大挑戰
- SQL腳本散落無管理：300個SQL檔案散落在Git倉庫各處，沒有版本控制、沒有依賴追蹤、沒有執行順序保證，需要dbt專案化管理和DAG自動解析
- 全量重新整理效能瓶頸：億級事實表每天全量重建耗時6小時，資源消耗巨大，需要增量模型只處理新增和變更資料
- 資料品質無人把關：上游欄位改名、空值激增、外鍵失效等問題在報表中才被發現，需要自動化資料測試和品質門禁
- 轉換邏輯重複編寫：日期維度產生、貨幣轉換、去重邏輯在每個模型中重複實作，需要巨集和套件來復用程式碼
- 生產部署缺乏規範：手動執行SQL、沒有CI/CD、沒有環境隔離、沒有回復機制，需要dbt Cloud或自動化流水線保障
分步實作：5種dbt資料建模模式
模式1：dbt專案搭建與模型設計
從零搭建dbt專案，定義資料來源、staging層和marts層模型。
# dbt_project.yml - 專案配置
name: analytics_warehouse
version: "1.0.0"
config-version: 2
profile: analytics_warehouse
model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]
target-path: "target"
clean-targets:
- "target"
- "dbt_packages"
models:
analytics_warehouse:
staging:
+materialized: view
+schema: staging
marts:
+materialized: table
+schema: marts
intermediate:
+materialized: ephemeral
# profiles.yml - 連線配置（~/.dbt/profiles.yml）
analytics_warehouse:
target: dev
outputs:
dev:
type: postgres
host: "{{ env_var('DBT_HOST') }}"
user: "{{ env_var('DBT_USER') }}"
password: "{{ env_var('DBT_PASSWORD') }}"
port: 5432
dbname: analytics_dev
schema: public
threads: 4
prod:
type: postgres
host: "{{ env_var('DBT_HOST') }}"
user: "{{ env_var('DBT_USER') }}"
password: "{{ env_var('DBT_PASSWORD') }}"
port: 5432
dbname: analytics_prod
schema: public
threads: 8
# models/staging/sources.yml - 資料來源宣告
version: 2
sources:
- name: raw_ecommerce
database: raw
schema: ecommerce
freshness:
warn_after: { count: 6, period: hour }
error_after: { count: 12, period: hour }
loaded_at_field: _etl_loaded_at
tables:
- name: orders
description: "原始訂單表"
columns:
- name: order_id
description: "訂單唯一ID"
tests:
- unique
- not_null
- name: customer_id
description: "客戶ID"
- name: order_status
description: "訂單狀態"
- name: order_total
description: "訂單金額"
- name: created_at
description: "建立時間"
- name: customers
description: "原始客戶表"
columns:
- name: customer_id
tests:
- unique
- not_null
- name: email
tests:
- unique
- name: order_items
description: "原始訂單明細表"
- name: products
description: "原始商品表"
-- models/staging/stg_orders.sql
-- Staging層：清洗和標準化原始資料
with source as (
select * from {{ source('raw_ecommerce', 'orders') }}
),
renamed as (
select
order_id,
customer_id,
order_status,
order_total::numeric(12,2) as order_total,
created_at as order_created_at,
_etl_loaded_at
from source
where order_id is not null
and order_status in ('pending', 'processing', 'shipped', 'delivered', 'cancelled')
),
final as (
select
*,
date_trunc('day', order_created_at) as order_date
from renamed
)
select * from final
-- models/staging/stg_customers.sql
with source as (
select * from {{ source('raw_ecommerce', 'customers') }}
),
renamed as (
select
customer_id,
trim(email) as email,
trim(first_name) as first_name,
trim(last_name) as last_name,
created_at as customer_created_at
from source
where customer_id is not null
)
select * from renamed
-- models/marts/fct_orders.sql
-- Marts層：面向業務的事實表
with orders as (
select * from {{ ref('stg_orders') }}
),
customers as (
select * from {{ ref('stg_customers') }}
),
order_items as (
select * from {{ ref('stg_order_items') }}
),
enriched as (
select
o.order_id,
o.customer_id,
c.email as customer_email,
o.order_status,
o.order_total,
o.order_date,
count(oi.order_item_id) as item_count,
coalesce(sum(oi.quantity), 0) as total_quantity
from orders o
left join customers c on o.customer_id = c.customer_id
left join order_items oi on o.order_id = oi.order_id
group by 1, 2, 3, 4, 5, 6
)
select * from enriched
-- models/marts/dim_customers.sql
-- Marts層：維度表
with customers as (
select * from {{ ref('stg_customers') }}
),
orders as (
select * from {{ ref('stg_orders') }}
),
customer_order_summary as (
select
customer_id,
count(*) as lifetime_order_count,
coalesce(sum(order_total), 0) as lifetime_order_value,
min(order_date) as first_order_date,
max(order_date) as last_order_date
from orders
where order_status != 'cancelled'
group by 1
),
final as (
select
c.customer_id,
c.email,
c.first_name,
c.last_name,
c.customer_created_at,
coalesce(cos.lifetime_order_count, 0) as lifetime_order_count,
coalesce(cos.lifetime_order_value, 0) as lifetime_order_value,
cos.first_order_date,
cos.last_order_date,
case
when cos.lifetime_order_count is null then 'prospect'
when cos.lifetime_order_count = 1 then 'one_time_buyer'
when cos.lifetime_order_count between 2 and 5 then 'repeat_buyer'
else 'vip'
end as customer_segment
from customers c
left join customer_order_summary cos on c.customer_id = cos.customer_id
)
select * from final
# 初始化專案並執行
pip install dbt-postgres==1.8.6
dbt init analytics_warehouse
dbt debug # 驗證連線
dbt run --select staging # 執行staging層模型
dbt run --select marts # 執行marts層模型
dbt run --full-refresh # 全量重新整理
dbt ls --resource-type model # 列出所有模型
dbt compile # 僅編譯不執行
模式2：增量模型與快照
增量模型只處理新增資料，快照追蹤維度表歷史變更。
-- models/staging/stg_order_events.sql
-- 增量模型：只處理新增事件資料
{{ config(
materialized='incremental',
unique_key='event_id',
incremental_strategy='merge',
on_schema_change='append_new_columns'
) }}
with source as (
select * from {{ source('raw_ecommerce', 'order_events') }}
),
filtered as (
select
event_id,
order_id,
event_type,
event_timestamp,
payload
from source
{% if is_incremental() %}
-- 增量模式：只處理上次執行之後的新資料
where event_timestamp > (select max(event_timestamp) from {{ this }})
{% endif %}
),
deduplicated as (
select
*,
row_number() over (
partition by event_id
order by event_timestamp desc
) as row_num
from filtered
)
select
event_id,
order_id,
event_type,
event_timestamp,
payload
from deduplicated
where row_num = 1
-- models/marts/fct_daily_order_metrics.sql
-- 增量彙總：每日訂單指標
{{ config(
materialized='incremental',
unique_key='metric_date',
incremental_strategy='delete+insert'
) }}
with orders as (
select * from {{ ref('stg_orders') }}
{% if is_incremental() %}
where order_date >= (select date(max(metric_date)) from {{ this }})
{% endif %}
),
daily_metrics as (
select
order_date as metric_date,
count(*) as total_orders,
count(*) filter (where order_status = 'delivered') as delivered_orders,
count(*) filter (where order_status = 'cancelled') as cancelled_orders,
sum(order_total) as total_revenue,
avg(order_total) as avg_order_value,
count(distinct customer_id) as unique_customers
from orders
group by 1
)
select * from daily_metrics
-- snapshots/customers_snapshot.sql
-- 快照：追蹤客戶維度表變更（SCD Type 2）
{% snapshot customers_snapshot %}
{{
config(
target_schema='snapshots',
strategy='timestamp',
unique_key='customer_id',
updated_at='customer_updated_at',
invalidate_hard_deletes=True
)
}}
select * from {{ source('raw_ecommerce', 'customers') }}
{% endsnapshot %}
-- snapshots/products_snapshot.sql
-- 快照：使用check策略追蹤商品變更
{% snapshot products_snapshot %}
{{
config(
target_schema='snapshots',
strategy='check',
unique_key='product_id',
check_cols=['product_name', 'category', 'price', 'status'],
invalidate_hard_deletes=True
)
}}
select * from {{ source('raw_ecommerce', 'products') }}
{% endsnapshot %}
# models/staging/stg_order_pivot.py
# dbt 1.8+ Python模型：使用Pandas進行複雜轉換
import pandas as pd
def model(dbt, session):
dbt.config(
materialized="table",
packages=["pandas==2.2.0"]
)
# 讀取上游dbt模型
order_events_df = dbt.ref("stg_order_events")
# 轉為Pandas DataFrame
if hasattr(order_events_df, 'to_pandas'):
df = order_events_df.to_pandas()
else:
df = order_events_df
# 透視表：每個訂單的事件時間線
pivot_df = df.pivot_table(
index='order_id',
columns='event_type',
values='event_timestamp',
aggfunc='min'
).reset_index()
pivot_df.columns = [f'first_{col}_at' if col != 'order_id' else col
for col in pivot_df.columns]
return pivot_df
# 執行增量模型和快照
dbt run --select stg_order_events # 首次全量
dbt run --select stg_order_events # 第二次增量
dbt snapshot # 執行快照
dbt snapshot --select customers_snapshot # 指定快照
dbt run --select stg_order_pivot # Python模型
模式3：dbt測試與資料品質
資料測試是dbt的核心價值——自動化斷言資料品質規則。
# models/marts/marts_schema.yml
version: 2
models:
- name: fct_orders
description: "訂單事實表，包含訂單維度和度量"
columns:
- name: order_id
description: "訂單唯一ID"
tests:
- unique
- not_null
- name: customer_id
tests:
- not_null
- relationships:
to: ref('dim_customers')
field: customer_id
- name: order_total
tests:
- not_null
- dbt_utils.accepted_range:
min_value: 0
max_value: 1000000
- name: order_status
tests:
- accepted_values:
values: ['pending', 'processing', 'shipped', 'delivered', 'cancelled']
- name: item_count
tests:
- not_null
- dbt_utils.accepted_range:
min_value: 0
- name: dim_customers
description: "客戶維度表"
columns:
- name: customer_id
tests:
- unique
- not_null
- name: email
tests:
- unique
- not_null
- name: customer_segment
tests:
- accepted_values:
values: ['prospect', 'one_time_buyer', 'repeat_buyer', 'vip']
-- tests/test_order_total_positive.sql
-- 自訂Singular測試：訂單金額必須為正數
select
order_id,
order_total
from {{ ref('fct_orders') }}
where order_total < 0
-- tests/test_fresh_order_delivery.sql
-- 自訂測試：已出貨訂單必須在7天內送達
select
o.order_id,
o.order_date,
d.first_delivered_at
from {{ ref('fct_orders') }} o
left join {{ ref('stg_order_pivot') }} d on o.order_id = d.order_id
where o.order_status = 'delivered'
and d.first_delivered_at is not null
and d.first_delivered_at::date > o.order_date + interval '7 days'
-- macros/test_at_least_one.sql
-- 參數化測試巨集：斷言表至少有N列
{% test at_least_one(model, column_name, min_count=1) %}
select {{ column_name }}
from {{ model }}
having count(*) < {{ min_count }}
{% endtest %}
# 在模型中使用自訂測試
models:
- name: fct_daily_order_metrics
tests:
- at_least_one:
column_name: metric_date
min_count: 1
columns:
- name: total_revenue
tests:
- not_null
- dbt_utils.accepted_range:
min_value: 0
# tests/test_customer_segment_distribution.py
# dbt 1.8+ Python測試
import pytest
def test_customer_segment_distribution(dbt):
"""驗證客戶分段的分布合理性"""
result = dbt.run_query("""
select
customer_segment,
count(*) as cnt,
count(*) * 100.0 / sum(count(*)) over() as pct
from {{ ref('dim_customers') }}
group by 1
order by 2 desc
""")
for row in result:
pct = float(row['pct'])
# 任何分段不應超過80%
assert pct < 80, f"Segment {row['customer_segment']} dominates: {pct:.1f}%"
# 執行測試
dbt test # 執行所有測試
dbt test --select fct_orders # 測試指定模型
dbt test --select at_least_one # 測試指定測試名稱
dbt test --exclude sensitive # 排除特定標籤
dbt test --store-failures # 儲存失敗記錄到表中
dbt test --select tag:critical # 執行特定標籤的測試
dbt build # run + test 一步到位
模式4：巨集與自訂物化
巨集是dbt的程式碼復用機制，自訂物化擴展dbt的核心能力。
-- macros/generate_date_dimension.sql
-- 產生日期維度表
{% macro generate_date_dimension(start_date, end_date) %}
with date_spine as (
{{ dbt_utils.date_spine(
datepart="day",
start_date="'" ~ start_date ~ "'::date",
end_date="'" ~ end_date ~ "'::date"
) }}
),
enriched as (
select
date_day as date_key,
date_day::date as full_date,
extract(year from date_day) as year_number,
extract(quarter from date_day) as quarter_number,
extract(month from date_day) as month_number,
extract(week from date_day) as week_number,
extract(day from date_day) as day_number,
extract(dow from date_day) as day_of_week,
to_char(date_day, 'Month') as month_name,
to_char(date_day, 'Day') as day_name,
extract(year from date_day) * 100 + extract(quarter from date_day)::int as year_quarter,
extract(year from date_day) * 100 + extract(month from date_day)::int as year_month,
case
when extract(dow from date_day) in (0, 6) then true
else false
end as is_weekend,
case
when extract(month from date_day) in (11, 12) then true
else false
end as is_holiday_season
from date_spine
)
select * from enriched
{% endmacro %}
-- models/marts/dim_date.sql
-- 使用日期維度巨集
{{ config(materialized='table') }}
{{ generate_date_dimension('2020-01-01', '2030-12-31') }}
-- macros/currency_conversion.sql
-- 貨幣轉換巨集
{% macro convert_currency(amount, from_currency, target_currency='USD', rate_date='current_date') %}
{% if from_currency == target_currency %}
{{ amount }}
{% else %}
(
{{ amount }}
* (
select exchange_rate
from {{ ref('stg_exchange_rates') }}
where from_currency = '{{ from_currency }}'
and to_currency = '{{ target_currency }}'
and rate_date = {{ rate_date }}
limit 1
)
)
{% endif %}
{% endmacro %}
-- macros/deduplicate.sql
-- 通用去重巨集
{% macro deduplicate(model, partition_by, order_by) %}
with ranked as (
select
*,
row_number() over (
partition by {{ partition_by }}
order by {{ order_by }}
) as _row_num
from {{ model }}
)
select *
from ranked
where _row_num = 1
{% endmacro %}
-- models/staging/stg_payments.sql
-- 使用貨幣轉換巨集
{{ config(materialized='incremental', unique_key='payment_id') }}
with source as (
select * from {{ source('raw_ecommerce', 'payments') }}
),
converted as (
select
payment_id,
order_id,
payment_method,
{{ convert_currency('amount', 'currency_code', 'USD') }} as amount_usd,
payment_status,
created_at
from source
{% if is_incremental() %}
where created_at > (select max(created_at) from {{ this }})
{% endif %}
)
select * from converted
-- macros/materializations/insert_overwrite.sql
-- 自訂物化：Insert Overwrite（適用於分割表）
{% materialization insert_overwrite, default %}
{%- set unique_key = config.get('unique_key') -%}
{%- set partition_by = config.get('partition_by') -%}
{%- set target_relation = this -%}
{%- set tmp_relation = make_temp_relation(target_relation) -%}
-- 編譯模型SQL
{% set sql = compiled_code %}
-- 建立臨時表
{{ create_table_as(True, tmp_relation, sql) }}
-- 刪除目標分割區資料
{% if partition_by %}
delete from {{ target_relation }}
where {{ partition_by }} in (
select distinct {{ partition_by }} from {{ tmp_relation }}
);
{% endif %}
-- 插入新資料
insert into {{ target_relation }}
select * from {{ tmp_relation }};
{{ return({'relations': [target_relation]}) }}
{% endmaterialization %}
# 執行和使用巨集
dbt run --select dim_date # 日期維度
dbt run --select stg_payments # 貨幣轉換
dbt run-operation generate_date_dimension --args '{"start_date": "2024-01-01", "end_date": "2026-12-31"}'
模式5：dbt Cloud CI/CD與生產部署
從開發到生產的完整CI/CD流水線。
# .github/workflows/dbt_ci_cd.yml
name: dbt CI/CD Pipeline
on:
pull_request:
branches: [main]
push:
branches: [main]
env:
DBT_HOST: ${{ secrets.DBT_HOST }}
DBT_USER: ${{ secrets.DBT_USER }}
DBT_PASSWORD: ${{ secrets.DBT_PASSWORD }}
jobs:
dbt-ci:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install dbt
run: pip install dbt-postgres==1.8.6 dbt-utils==1.3.0
- name: dbt Debug
run: dbt debug --target dev
- name: dbt Compile
run: dbt compile --target dev
- name: dbt Test (Staging)
run: dbt test --select staging --target dev
- name: dbt Build (PR Check)
if: github.event_name == 'pull_request'
run: |
dbt build --target dev --full-refresh 2>&1 | tee build_output.txt
echo "## dbt Build Summary" >> $GITHUB_STEP_SUMMARY
echo '```' >> $GITHUB_STEP_SUMMARY
tail -20 build_output.txt >> $GITHUB_STEP_SUMMARY
echo '```' >> $GITHUB_STEP_SUMMARY
dbt-deploy:
needs: dbt-ci
if: github.ref == 'refs/heads/main'
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install dbt
run: pip install dbt-postgres==1.8.6 dbt-utils==1.3.0
- name: dbt Build (Production)
run: dbt build --target prod
- name: dbt Snapshot (Production)
run: dbt snapshot --target prod
- name: dbt Source Freshness
run: dbt source freshness --target prod
- name: Generate Docs
run: dbt docs generate --target prod
- name: Deploy Docs
uses: peaceiris/actions-gh-pages@v4
with:
github_token: ${{ secrets.GITHUB_TOKEN }}
publish_dir: ./target
# dbt Cloud部署配置（dbt_cloud.yml）
# dbt Cloud Job配置
name: Production Daily Build
schedule:
cron: "0 6 * * *"
target: prod
steps:
- dbt source freshness
- dbt build --full-refresh
- dbt snapshot
- dbt docs generate
notifications:
- on_failure:
channels:
- slack:#data-alerts
- email:data-team@company.com
- on_success:
channels:
- slack:#data-updates
# 生產部署命令
dbt build --target prod --full-refresh # 全量建構
dbt build --target prod --select staging # 只建構staging層
dbt build --target prod --select +fct_orders # 建構fct_orders及其上游
dbt build --target prod --select fct_orders+ # 建構fct_orders及其下游
dbt source freshness --target prod # 檢查資料新鮮度
dbt docs generate --target prod # 產生文件
dbt run --target prod --select state:modified # 只執行修改過的模型（狀態對比）
dbt build --target prod --defer --state ./prod_state # 延遲解析
# scripts/dbt_deploy_check.py
# 部署前檢查腳本
import subprocess
import sys
def run_dbt_command(cmd: str, target: str = "prod") -> bool:
"""執行dbt命令並檢查結果"""
full_cmd = f"dbt {cmd} --target {target}"
result = subprocess.run(full_cmd.split(), capture_output=True, text=True)
if result.returncode != 0:
print(f"FAILED: {full_cmd}")
print(result.stderr)
return False
print(f"PASSED: {full_cmd}")
return True
def main():
checks = [
("debug", "dev"),
("compile", "dev"),
("test --select tag:critical", "dev"),
("source freshness", "prod"),
]
all_passed = True
for cmd, target in checks:
if not run_dbt_command(cmd, target):
all_passed = False
if not all_passed:
print("Pre-deploy checks FAILED. Aborting deployment.")
sys.exit(1)
print("All pre-deploy checks PASSED. Proceeding with deployment.")
run_dbt_command("build --full-refresh", "prod")
run_dbt_command("snapshot", "prod")
run_dbt_command("docs generate", "prod")
if __name__ == "__main__":
main()
5大避坑指南
1. 在增量模型中忘記is_incremental()條件
❌ 錯誤做法：增量模型沒有增量過濾條件，每次都全量處理
-- ❌ 沒有增量過濾，每次全量
{{ config(materialized='incremental', unique_key='id') }}
select * from {{ source('raw', 'events') }}
✅ 正確做法：使用is_incremental()巨集只處理新增資料
-- ✅ 增量過濾
{{ config(materialized='incremental', unique_key='id') }}
select * from {{ source('raw', 'events') }}
{% if is_incremental() %}
where created_at > (select max(created_at) from {{ this }})
{% endif %}
2. 在模型中硬編碼資料來源表名
❌ 錯誤做法：直接寫表名，dbt無法追蹤依賴和血緣
-- ❌ 硬編碼表名
select * from raw.ecommerce.orders
✅ 正確做法：使用source()和ref()函式
-- ✅ 使用source和ref
select * from {{ source('raw_ecommerce', 'orders') }}
select * from {{ ref('stg_orders') }}
3. 物化策略選擇不當
❌ 錯誤做法：所有模型都用table物化，大資料量全量重建耗時巨大
✅ 正確做法：staging層用view，marts事實表用incremental，小維度表用table，中間轉換用ephemeral
models:
analytics_warehouse:
staging:
+materialized: view # 輕量，不儲存資料
marts:
+materialized: table # 事實表可改為incremental
intermediate:
+materialized: ephemeral # 內聯CTE，不建立物件
4. 忽略快照策略選擇
❌ 錯誤做法：所有快照都用timestamp策略，但上游沒有可靠的更新時間欄位
✅ 正確做法：有可靠updated_at欄位用timestamp策略，否則用check策略追蹤指定欄變更
-- ✅ 有updated_at用timestamp
{% snapshot customers_snapshot %}
{{ config(strategy='timestamp', updated_at='customer_updated_at') }}
select * from {{ source('raw_ecommerce', 'customers') }}
{% endsnapshot %}
-- ✅ 沒有updated_at用check
{% snapshot products_snapshot %}
{{ config(strategy='check', check_cols=['name', 'price', 'category']) }}
select * from {{ source('raw_ecommerce', 'products') }}
{% endsnapshot %}
5. CI/CD中缺少資料測試門禁
❌ 錯誤做法：CI只執行dbt run，不執行dbt test，資料品質問題直接進入生產
✅ 正確做法：使用dbt build（run + test），關鍵測試失敗則阻斷部署
# ✅ CI中使用dbt build
- name: dbt Build with Tests
run: dbt build --target dev
- name: Critical Tests Gate
run: dbt test --select tag:critical --target dev
10大報錯排查
| 序號 | 報錯資訊 | 原因 | 解決方法 |
|---|---|---|---|
| 1 | Compilation Error: 'source' takes exactly two arguments |
source()函式參數錯誤 |
檢查source('schema_name', 'table_name')格式，確保sources.yml中已定義 |
| 2 | Database Error: relation "stg_xxx" does not exist |
ref()參照的模型尚未執行 |
先執行上游模型：dbt run --select stg_xxx；檢查模型檔案名稱和ref名稱一致 |
| 3 | Compilation Error: dbt was unable to find a matching resource |
ref/source參照的資源不存在 | 執行dbt ls檢查資源列表；檢查YAML中的名稱和SQL中的參照是否一致 |
| 4 | Incremental model running full refresh unexpectedly |
增量模型首次執行會全量重新整理 | 首次執行是正常的全量；後續檢查unique_key是否匹配 |
| 5 | Snapshot detected a change but dbt_snapshot_check is missing |
check策略的快照缺少check_cols配置 | 在snapshot配置中明確指定check_cols列表 |
| 6 | Test failed: accepted_values - got 1 unexpected value |
欄中存在未在accepted_values中列出的值 | 檢查資料中的實際值：select distinct col from model |
| 7 | Runtime Error: could not connect to server |
資料庫連線失敗 | 執行dbt debug檢查連線；驗證環境變數和profiles.yml配置 |
| 8 | Compilation Error: macro 'xxx' not found |
巨集檔案路徑錯誤或未安裝依賴套件 | 檢查macro-paths配置；安裝缺少的套件：dbt deps |
| 9 | Schema Error: on_schema_change='append_new_columns' failed |
增量模型新增欄但目標表已存在資料 | 使用on_schema_change='append_new_columns'或--full-refresh重建 |
| 10 | dbt source freshness: source is stale |
資料來源freshness檢查失敗 | 檢查上游ETL是否正常；調整freshness閾值；檢查loaded_at_field配置 |
進階最佳化技巧
1. 狀態對比建構
利用dbt的state機制，只執行修改過的模型及其下游，大幅減少CI建構時間。
# 產生生產狀態artifact
dbt build --target prod
cp target/manifest.json ./prod_manifest.json
# CI中只建構變更的模型
dbt build --select state:modified --state ./prod_state --defer
2. 分層平行執行
利用dbt的執行緒配置和模型依賴自動平行執行無依賴的模型。
# profiles.yml中配置執行緒數
analytics_warehouse:
outputs:
prod:
threads: 8 # 8個平行執行緒
# 按層平行執行
dbt run --select staging & # 背景執行staging
dbt run --select marts # marts依賴staging，會等待
3. 模型標籤與選擇性執行
# models/staging/staging_schema.yml
models:
- name: stg_orders
config:
tags: ['critical', 'revenue']
- name: stg_customers
config:
tags: ['critical']
- name: stg_page_views
config:
tags: ['experimental']
dbt build --select tag:critical # 只建構關鍵模型
dbt test --select tag:experimental # 只測試實驗模型
dbt run --select tag:revenue # 只執行收入相關
4. dbt-utils和dbt-expectations套件
# packages.yml
packages:
- package: dbt-labs/dbt_utils
version: "1.3.0"
- package: calogica/dbt_expectations
version: "0.10.0"
- package: dbt-labs/dbt_audit_helper
version: "0.12.0"
-- 使用dbt-utils巨集
{{ dbt_utils.date_spine('day', "'2024-01-01'::date", "'2026-12-31'::date") }}
{{ dbt_utils.pivot('payment_method', dbt_utils.get_column_values(ref('stg_payments'), 'payment_method')) }}
{{ dbt_utils.safe_divide('numerator', 'denominator') }}
5. 資料血緣與文件自動產生
# 產生並檢視文件
dbt docs generate
dbt docs serve # 本地啟動文件服務
# 產生CI/CD中的資料血緣報告
dbt compile --target prod
dbt docs generate --target prod
# models/marts/marts_docs.md
{% docs order_total %}
訂單總金額，單位為USD。已取消的訂單金額為0。
{% enddocs %}
{% docs customer_segment %}
客戶生命週期分段：
- prospect: 註冊但未下單
- one_time_buyer: 僅1筆訂單
- repeat_buyer: 2-5筆訂單
- vip: 超過5筆訂單
{% enddocs %}
資料轉換工具對比
| 維度 | dbt | Dataform | SQLMesh | 預存程序 |
|---|---|---|---|---|
| 核心理念 | Transform in warehouse | Transform in warehouse (GCP) | Transform with versioning | Transform in database |
| 語言 | SQL + Jinja + Python | SQLX + JavaScript | SQL + Python | SQL |
| 版本管理 | Git原生 | Git原生 | Git原生 + 虛擬環境 | 手動腳本 |
| 增量模型 | 原生支援，策略豐富 | 原生支援 | 原生支援，自動推斷 | 手動實作 |
| 資料測試 | 原生支援，4種內建+自訂 | 原生支援 | 原生支援 | 手動實作 |
| 資料血緣 | 自動產生 | 自動產生 | 自動產生 | 無 |
| 文件產生 | 自動產生 | 自動產生 | 自動產生 | 無 |
| CI/CD整合 | dbt Cloud + GitHub Actions | GCP原生 | 原生CI | 手動 |
| 雲平台 | 全平台 | GCP優先 | 全平台 | 資料庫綁定 |
| 社群生態 | 最成熟，套件最多 | GCP生態 | 增長中 | 無 |
| 學習曲線 | 中等 | 低 | 中高 | 低 |
| 2026推薦度 | ★★★★★ | ★★★★ | ★★★★ | ★★ |
| 適用場景 | 通用資料轉換 | GCP使用者 | 版本化需求強 | 簡單轉換 |
相關工具
- JSON格式化 — 格式化dbt模型YAML配置和測試結果JSON
- 雜湊加密 — 產生資料指紋用於增量模型去重和快照校驗
- cURL轉程式碼 — 將資料來源API除錯的cURL命令轉為Python程式碼
延伸閱讀
- Python AI資料流水線實戰 — 從ETL到特徵儲存的完整資料流水線
- Apache Airflow資料管道實戰 — dbt模型的上游編排排程
- PostgreSQL連線池最佳化 — dbt背後的資料庫連線最佳化
- Python Pydantic V2資料校驗 — 資料管道中的Schema校驗防線
dbt在2026年已成為現代資料棧轉換層的事實標準。記住五個核心原則：專案化管理替代散落SQL——一個dbt專案統一管理所有轉換邏輯，依賴自動解析；增量模型是大資料量的生命線——
is_incremental()配合unique_key，只處理新增資料；資料測試是品質的底線——unique、not_null、relationships三件套加上自訂測試，上線前必須全綠；巨集和套件是復用的基石——一次編寫，全團隊復用，告別複製貼上；CI/CD是生產的保障——從PR檢查到生產部署，每一步都有自動化門禁。從專案搭建到生產運維，dbt的生態已經足夠成熟，剩下的就是你在業務場景中的實踐了。
本站提供瀏覽器本地工具，免註冊即可試用 →
#dbt数据转换#数据建模#SQL转换#dbt Core#2026#AI与大数据