dbt 与 DataOps

本章目标：能搭一个最小 dbt 项目，用 ref 组织依赖、用测试约束数据质量、用增量模型降低重跑成本，并说清上线流程。 前置知识：数仓分层、SQL 建表与增删改 预计用时：110 分钟

手写 SQL 任务时，表一多就不知道谁依赖谁；改一个字段，下游全挂；数据错了，只能靠人肉对比。dbt 把 SQL 当代码管理，补上依赖、测试、文档和版本控制。

1. 为什么需要 dbt

dbt 本身不存数据，它把 SQL 编译成 CREATE TABLE AS、INSERT、MERGE，交给底层数据库执行。数据平台可以是 DuckDB、Snowflake、BigQuery、MaxCompute（通过适配器）。

2. 核心概念

2.1 项目结构

analytics/
  dbt_project.yml          项目配置
  profiles.yml             连接信息（不入库）
  seeds/orders_demo.csv    小规模静态数据
  models/staging/          stg_orders.sql 清洗，schema.yml 测试
  models/marts/            ads_city_gmv.sql、fact_order_inc.sql
  snapshots/               orders_snapshot.sql

staging 贴近源表，只做清洗；marts 面向业务，做聚合和宽表。

2.2 ref 与依赖

不要在模型里写死表名：

SELECT * FROM analytics.staging.stg_orders

改用：

SELECT * FROM {{ ref('stg_orders') }}

ref() 会解析成当前环境的真实表名，并让 dbt 知道 ads_city_gmv 依赖 stg_orders。执行时 dbt 自动按依赖顺序跑，不需要手工排任务。

2.3 物化方式

2.4 测试

• 通用测试：在 YAML 里声明 unique、not_null、accepted_values、relationships。
• 自定义测试：写一段 SQL，返回的行就是失败记录。

测试失败会让 dbt build 中断，避免坏数据流向下游。

2.5 DataOps 的四个动作

开发：分支 + 本地 dbt build
评审：看 SQL diff、测试结果、血缘影响
发布：CI 跑全量构建，合并后调度执行
运行：监控失败、数据量异常、数据新鲜度

DataOps 不是某个工具，而是把“改 SQL”变成和“改代码”一样的受控流程。

3. 动手示例

下面从零搭一个能跑的项目，使用 dbt-duckdb，不需要数据库服务器。

3.1 安装与准备

pip install dbt-duckdb
mkdir -p analytics/seeds analytics/models/staging analytics/models/marts analytics/snapshots
cp data/sample/orders_demo.csv analytics/seeds/orders_demo.csv

3.2 项目配置

analytics/dbt_project.yml：

name: analytics
version: "1.0.0"
config-version: 2
profile: analytics
model-paths: ["models"]
seed-paths: ["seeds"]
snapshot-paths: ["snapshots"]
models:
  analytics:
    staging: {+materialized: view}
    marts: {+materialized: table}

analytics/profiles.yml：

analytics:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: "analytics/dev.duckdb"
      threads: 4

连接文件可能包含密码，不要提交到 Git；用环境变量或 CI Secret 注入。

3.3 staging 模型与测试

analytics/models/staging/stg_orders.sql：

SELECT
    CAST(order_id AS VARCHAR) AS order_id,
    status,
    city,
    CAST(amount AS DECIMAL(12, 2)) AS amount,
    CAST(order_date AS DATE) AS order_date
FROM {{ ref('orders_demo') }}

analytics/models/staging/schema.yml：

version: 2
models:
  - name: stg_orders
    description: 订单清洗层，一行一个订单
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: ["paid", "canceled", "unavailable", "shipped"]
      - name: amount
        tests: [not_null]

stg_orders 物化成视图，不额外占存储；测试直接跑在视图上。

3.4 marts 聚合模型

analytics/models/marts/ads_city_gmv.sql：

SELECT
    city,
    SUM(amount) AS gmv,
    COUNT(*) AS order_cnt
FROM {{ ref('stg_orders') }}
WHERE status = 'paid'
GROUP BY city

运行后预期结果：北京 455（3 单）、上海 405（2 单）、广州 319（2 单）、深圳 180（1 单）。金额与订单数对不上时，先查 status 过滤和金额类型转换。

3.5 增量模型

订单事实表每天只处理新增数据。analytics/models/marts/fact_order_inc.sql：

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='delete+insert'
) }}

SELECT
    order_id,
    status,
    city,
    amount,
    order_date
FROM {{ ref('stg_orders') }}
{% if is_incremental() %}
WHERE order_date > (SELECT COALESCE(MAX(order_date), DATE '1900-01-01') FROM {{ this }})
{% endif %}

第一次跑建全表，之后只处理比已有最大日期更新的数据。delete+insert 保证同一主键重跑不会翻倍。实际业务中还要留出迟到数据窗口，例如回刷最近 3 天。

3.6 快照记录历史

analytics/snapshots/orders_snapshot.sql：

{% snapshot orders_snapshot %}
{{ config(
    target_schema='snapshots',
    unique_key='order_id',
    strategy='check',
    check_cols=['status']
) }}

SELECT * FROM {{ ref('stg_orders') }}
{% endsnapshot %}

status 变化会新增一条历史记录，并带上 dbt_valid_from、dbt_valid_to，等价于手工维护的拉链表。

3.7 构建与测试

P="--project-dir analytics --profiles-dir analytics"
dbt seed $P && dbt run $P && dbt test $P && dbt docs generate $P

dbt build 可以一次完成 seed、run、test、snapshot。任一测试失败，命令以非零状态退出，CI 会因此拦截合并。

3.8 用 CI 守住质量

.github/workflows/dbt-ci.yml：

name: dbt-ci
on: [pull_request]
jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install dbt-duckdb
      - run: dbt build --project-dir analytics --profiles-dir analytics

CI 用独立数据库或 DuckDB 临时文件，不连生产。合并后再由调度系统用生产连接执行。

4. 常见坑

4.1 把密码写进 profiles.yml

profiles.yml 可能被提交到仓库。用环境变量或平台的 Secret 管理，仓库里只留模板。

4.2 只跑 dbt run 不跑测试

模型能建出来不代表数据正确。主键重复、状态值拼错都能成功建表。CI 里要跑 dbt build 或至少 dbt test。

4.3 增量条件用 >=

WHERE order_date >= MAX(order_date) 会把最后一天重复处理。配合 delete+insert 可以幂等，但用 append 策略就会翻倍。先确认策略再写条件。

4.4 只测主键、直连生产、不看血缘

主键唯一只说明没有重复，不说明金额合理；补充非负、枚举值、表间关系测试。本地试错不要直连生产，用独立 schema 或独立数据库。一个 staging 字段改名可能影响十几个下游模型，发布前用 dbt ls --select +模型名 查下游。

4.5 快照当成每日全量备份

快照只在指定列变化时新增记录，不是每天的完整副本；需要每日状态快照，应单独建周期快照事实表。

5. 在线练习

6. 面试延伸

面试题：dbt 解决了什么问题？

是什么：把 SQL 模型纳入版本控制，用 ref 管理依赖，用测试保证质量，并自动生成文档和血缘。

解决什么问题：手写任务依赖靠人记、口径散落、改字段影响面不清。dbt 让 SQL 工程化，能评审、能回滚、能自动化。

我的实操边界：本章用 dbt-duckdb 在本地跑通 seed、run、test、snapshot，未在生产数据平台维护大型 dbt 项目。

面试题：view、table、incremental 怎么选？

是什么：view 不存数据，适合轻量清洗；table 全量重建，适合中小结果表；incremental 只处理增量，适合大事实表。

解决什么问题：全建 table 浪费存储和构建时间，全建 view 又会让查询反复计算。

我的实操边界：本章配置过三种物化方式，未做过大规模增量任务的性能调优。

7. 本章验收

• 能搭出包含 staging、marts、seeds 的 dbt 项目并成功运行。
• 能用 ref() 建立两个模型的依赖，并说清执行顺序。
• 能为订单表写出唯一、非空、枚举值三类测试。
• 能写出按日期增量、按主键幂等更新的模型。
• 能解释快照和每日全量备份的区别。
• 能画出从提交代码到调度运行的最小 DataOps 流程。