高级特性：packages / snapshots / seeds

8.1 dbt 包管理（Packages）

dbt 有一个活跃的包生态系统，可以从 hub.getdbt.com 浏览和安装社区包，这些包提供可复用的宏、通用测试、数据模型等。

YAML# packages.yml（项目根目录）
packages:
  # 官方工具宏包
  - package: dbt-labs/dbt_utils
    version: [">=1.2.0", "<2.0.0"]

  # 扩展测试断言包
  - package: calogica/dbt_expectations
    version: [">=0.10.0", "<0.11.0"]

  # 自动生成 staging 模型代码
  - package: dbt-labs/codegen
    version: [">=0.12.0", "<0.13.0"]

  # 数据审计：行数对比（旧版 vs 新版模型）
  - package: dbt-labs/audit_helper
    version: [">=0.10.0", "<0.11.0"]

  # 从 Git 仓库安装（自定义内部包）
  - git: "https://github.com/your-org/dbt-internal-macros.git"
    revision: v1.0.0

BASH# 安装 packages.yml 中声明的所有包
dbt deps

# 更新到最新兼容版本
dbt deps --upgrade

codegen 包：自动生成代码

BASH# 自动为 source 表生成 staging 模型 SQL
dbt run-operation codegen.generate_base_model \
  --args '{"source_name": "ecommerce", "table_name": "orders"}'

# 自动生成 schema.yml 中的列定义
dbt run-operation codegen.generate_model_yaml \
  --args '{"model_names": ["stg_orders", "stg_customers"]}'

8.2 Snapshots — SCD Type 2 历史快照

Snapshots 用于追踪数据随时间的变化，实现缓慢变化维度 Type 2（SCD Type 2）——保留每条记录的完整历史版本。

SQL-- snapshots/customers_snapshot.sql
{% snapshot customers_snapshot %}

{{ config(
    target_schema = 'snapshots',
    unique_key    = 'customer_id',
    strategy      = 'timestamp',         -- 或 'check'
    updated_at    = 'updated_at',
) }}

SELECT
    customer_id,
    email,
    phone,
    address,
    customer_tier,
    updated_at
FROM {{ source('ecommerce', 'customers') }}

{% endsnapshot %}

BASH# 运行 snapshots
dbt snapshot

快照表自动添加以下元数据列：

列名	说明
dbt_scd_id	每个快照行的唯一 ID（MD5 哈希）
dbt_updated_at	该版本记录的更新时间
dbt_valid_from	该版本记录的生效开始时间
dbt_valid_to	该版本记录的生效结束时间（NULL = 当前版本）

两种快照策略

SQL-- 策略 1：timestamp（推荐）
-- 当 updated_at 字段变化时，认为记录已更新
{{ config(
    strategy   = 'timestamp',
    updated_at = 'updated_at',
) }}

-- 策略 2：check（适合没有 updated_at 字段的表）
-- 当指定列的值发生变化时，认为记录已更新
{{ config(
    strategy     = 'check',
    check_cols   = ['email', 'address', 'customer_tier'],
) }}

8.3 Seeds — CSV 文件直接加载为表

Seeds 是放在 seeds/ 目录下的 CSV 文件，dbt 会将其加载为数据仓库中的表。适合小型的、相对静态的参考数据：

CSV">-- seeds/country_codes.csv
country_code,country_name,region
CN,China,Asia
US,United States,North America
DE,Germany,Europe
JP,Japan,Asia

CSV">-- seeds/product_categories.csv
category_id,category_name,parent_category
1,Electronics,
2,Phones,Electronics
3,Laptops,Electronics
4,Clothing,
5,Men's Wear,Clothing

BASH# 加载 seeds（CSV 文件）到数据仓库
dbt seed

# 只加载特定 seed
dbt seed --select country_codes

# 全量刷新（seed 数据有变更时使用）
dbt seed --full-refresh

SQL-- 在模型中引用 seed（与 ref() 引用模型相同）
SELECT
    o.*,
    cc.country_name,
    cc.region
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref('country_codes') }} cc
    ON o.shipping_country = cc.country_code

8.4 项目分层规范：staging / intermediate / marts

随着项目规模增长，规范的目录组织变得至关重要。社区最佳实践是三层架构：

TREE">models/
├── staging/                   # 第1层：1:1 清洗原始数据
│   ├── ecommerce/
│   │   ├── stg_orders.sql
│   │   ├── stg_customers.sql
│   │   └── schema.yml         # sources + staging 模型文档/测试
│   └── segment/
│       ├── stg_events.sql
│       └── schema.yml
│
├── intermediate/              # 第2层：中间计算（可选）
│   ├── int_orders_enriched.sql
│   ├── int_customer_segments.sql
│   └── schema.yml
│
└── marts/                     # 第3层：面向业务的最终表
    ├── core/
    │   ├── dim_customers.sql
    │   ├── fct_orders.sql
    │   └── schema.yml
    └── marketing/
        ├── fct_ad_performance.sql
        └── schema.yml

层级	命名规范	物化方式	原则
staging	stg_{source}_{table}	view	1:1 映射原始表，只做清洗（重命名、类型转换、去除无效行），不做业务逻辑
intermediate	int_{description}	view 或 ephemeral	跨源 join、复杂计算，不直接面向 BI 消费
marts	fct_{entity} / dim_{entity}	table 或 incremental	面向业务域（核心/营销/财务），事实表+维度表，直接供 BI 使用