dbt (data build tool)

dbt

dbt 讓分析工程師透過撰寫 SQL SELECT 語句來轉換資料。它處理實體化（表格、檢視、增量）、測試、文件與血緣追蹤。

安裝

# 安裝 dbt 搭配 PostgreSQL 適配器
pip install dbt-postgres

# 或搭配其他適配器
pip install dbt-bigquery
pip install dbt-snowflake

# 初始化新專案
dbt init my_project
cd my_project

專案結構

my_project/
├── dbt_project.yml      # 專案設定
├── profiles.yml         # 連線設定檔（通常在 ~/.dbt/）
├── models/
│   ├── staging/         # 原始資料清理
│   │   ├── _staging.yml # staging 模型的綱要與測試
│   │   ├── stg_users.sql
│   │   └── stg_orders.sql
│   └── marts/           # 商業邏輯
│       ├── _marts.yml
│       └── fct_revenue.sql
├── tests/               # 自訂資料測試
├── macros/              # 可重複使用的 SQL 巨集
└── seeds/               # 要載入的 CSV 檔案

設定

# dbt_project.yml：專案設定
name: my_project
version: '1.0.0'
profile: my_project

models:
  my_project:
    staging:
      +materialized: view
      +schema: staging
    marts:
      +materialized: table
      +schema: analytics

# profiles.yml：資料庫連線（~/.dbt/profiles.yml）
my_project:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: analyst
      password: "{{ env_var('DBT_PASSWORD') }}"
      dbname: analytics
      schema: dev
      threads: 4
    prod:
      type: postgres
      host: prod-db.example.com
      port: 5432
      user: dbt_prod
      password: "{{ env_var('DBT_PROD_PASSWORD') }}"
      dbname: analytics
      schema: public
      threads: 8

Staging 模型

-- models/staging/stg_users.sql：清理原始使用者資料
WITH source AS (
    SELECT * FROM {{ source('raw', 'users') }}
),

cleaned AS (
    SELECT
        id AS user_id,
        LOWER(TRIM(email)) AS email,
        name,
        created_at::timestamp AS signed_up_at,
        CASE WHEN status = 'active' THEN TRUE ELSE FALSE END AS is_active
    FROM source
    WHERE email IS NOT NULL
)

SELECT * FROM cleaned

-- models/staging/stg_orders.sql：清理原始訂單資料
SELECT
    id AS order_id,
    user_id,
    amount_cents / 100.0 AS amount,
    status,
    created_at::timestamp AS ordered_at
FROM {{ source('raw', 'orders') }}
WHERE status != 'test'

Mart 模型

-- models/marts/fct_revenue.sql：營收事實表
{{
    config(
        materialized='incremental',
        unique_key='order_date',
        on_schema_change='sync_all_columns'
    )
}}

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
    {% if is_incremental() %}
    WHERE ordered_at > (SELECT MAX(order_date) FROM {{ this }})
    {% endif %}
),

daily AS (
    SELECT
        DATE_TRUNC('day', ordered_at)::date AS order_date,
        COUNT(*) AS total_orders,
        COUNT(DISTINCT user_id) AS unique_customers,
        SUM(amount) AS total_revenue,
        AVG(amount) AS avg_order_value
    FROM orders
    WHERE status = 'completed'
    GROUP BY 1
)

SELECT * FROM daily

綱要與測試

# models/staging/_staging.yml：定義來源、欄位與測試
version: 2

sources:
  - name: raw
    schema: public
    tables:
      - name: users
        loaded_at_field: created_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}
      - name: orders

models:
  - name: stg_users
    description: 清理後的使用者資料
    columns:
      - name: user_id
        tests: [unique, not_null]
      - name: email
        tests: [unique, not_null]

  - name: stg_orders
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'completed', 'cancelled', 'refunded']

CLI 指令

# commands.sh：常用 dbt CLI 指令
# 執行所有模型
dbt run

# 執行特定模型及其上游依賴
dbt run --select +fct_revenue

# 執行測試
dbt test

# 產生並提供文件
dbt docs generate
dbt docs serve --port 8081

# 檢查來源新鮮度
dbt source freshness

# 完整建置（run + test + snapshot）
dbt build

# 針對正式環境執行
dbt run --target prod

巨集

-- macros/cents_to_dollars.sql：用於貨幣轉換的可重複使用巨集
{% macro cents_to_dollars(column_name) %}
    ({{ column_name }} / 100.0)::numeric(10,2)
{% endmacro %}

-- 在模型中使用：SELECT {{ cents_to_dollars('amount_cents') }} AS amount