dbt

dbt 是做什么的？

dbt 让分析工程师通过编写 SQL SELECT 语句来转换数据。它自动处理物化（表、视图、增量表）、测试、文档和血缘追踪。

安装

# 使用 PostgreSQL 适配器安装 dbt
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 模型的 schema 和测试
│   │   ├── 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

Schema 与测试

# 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

# 完整构建（运行 + 测试 + 快照）
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