dbt+SQLServer构建数据仓库(6)：Jinja与DAG实战

本文是第 6 篇,深入 dbt 的模板引擎和依赖图机制:Jinja 怎么把 {{ ref() }} 渲染成真实表名、DAG 怎么自动构建和排序、--select 选择器怎么精准控制要跑哪些模型。

一、引言

前 5 篇我们一直在写 {{ ref('xxx') }} 和 {{ source('raw','xxx') }},但只是把它们当成"魔法字符串"用——能跑就行,没追问它背后到底干了什么。这一篇就钻进 dbt 的两个底层机制:

• Jinja 模板引擎:让 SQL 变成"可编程"的代码,{{ }} 里的表达式在编译期被替换成真实表名。
• DAG 依赖图:dbt 扫描所有 .sql 里的 ref()/source(),自动构建一张有向无环图,据此决定执行顺序。

理解了这两件事,你才能用好 --select 选择器精准控制跑哪些模型,才能在编译报错时一眼看出是模板渲染问题还是依赖问题,后面写 macro 也是同一套套路。

二、Jinja 基础:dbt 的模板引擎

Jinja 本来是 Python 生态里一个通用模板引擎(Flask 的页面模板就是它),dbt 把它搬到了 SQL 上,让纯文本的 SQL 文件变成"可编程"的模板。

两种标记

标记	作用	例子
{{ ... }}	渲染表达式:变量值、函数调用的返回值会被原地替换进 SQL	{{ ref('stg_customers') }}
{% ... %}	执行语句:控制流(if/for)、赋值(set),不直接产出文本	{% set x = 1 %}、{% if ... %}

另外 {# ... #} 是注释,渲染后不会出现在编译结果里。

dbt 给 Jinja 加的"私货"

Jinja 本身只有基本控制流,dbt 往上下文里塞了几十个内置函数和对象,常用的有:

• ref('model_name') —— 引用项目内的另一个 model,同时声明依赖
• source('source_name','table_name') —— 引用项目外的源表
• config(materialized='table') —— 在模型内动态配置物化方式
• var('key', default) —— 读 dbt_project.yml / vars 里的变量
• env_var('PATH') —— 读宿主机环境变量

本项目的用法

本项目刻意把 Jinja 用得很克制,基本只用到 {{ ref() }} 和 {{ source() }},没写自定义 macro,也没用循环/条件。维护成本低,但底层机制是一样的——下面两节就拆开看这两个函数到底干了什么。

三、ref() 深度解析

案例代码

[models/marts/dim_customers.sql] 里出现了 4 次 ref():

with customers as (
    select * from {{ ref('stg_customers') }}
),
orders as (
    select
        customer_id,
        min(order_date) as first_order_date,
        max(order_date) as most_recent_order_date,
        count(order_id) as number_of_orders
    from {{ ref('stg_orders') }}
    group by customer_id
),
payments as (
    select
        o.customer_id,
        sum(p.amount) as lifetime_value
    from {{ ref('stg_orders') }} o
    inner join {{ ref('stg_payments') }} p
        on o.order_id = p.order_id
    where p.status = 'completed'
    group by o.customer_id
)
select ...
from customers c
left join orders o on c.customer_id = o.customer_id
left join payments p on c.customer_id = p.customer_id

ref() 干了两件事

1. 声明依赖

dbt 在解析阶段(不是执行阶段)扫到 {{ ref('stg_customers') }},就在内部 DAG 里加一条边:stg_customers → dim_customers。这就是为什么 dbt 知道 dim_customers 要在 stg_customers 之后跑——它根本不是去猜,而是你用 ref() 显式告诉它的。

2. 解析表名

到了编译期,ref() 把 'stg_customers' 这个逻辑名翻译成数据库里的真实表名,规则大致是:

最终 schema = target.schema + custom_schema (custom_schema 来自 dbt_project.yml 的 models: 配置)
最终表名   = model name (即 .sql 文件名去掉扩展名)

本项目 target.schema = dbt_dev,staging 层配了 custom_schema: staging,所以 staging 模型落在 dbt_dev_staging schema;marts 层没配 custom_schema,直接落在 dbt_dev schema。

编译前后对比

	代码
编译前(你写的)	from {{ ref('stg_customers') }}
编译后(dbt 生成)	from [dbt_dev_staging].[stg_customers]

注意 SQL Server 方言用 方括号 [...] 包标识符,而不是 PostgreSQL 的双引号 "..."。这是 dbt-sqlserver adapter 干的活,你不用管。

ref() vs 硬编码表名

方式	写法	依赖追踪	环境隔离	改名成本
ref()	{{ ref('stg_customers') }}	自动	自动(dev/prod 不同 schema)	只改一处
硬编码	from dbt_dev_staging.stg_customers	无	手动改环境前缀	全文搜索替换

一句话:用 ref() 你写的是逻辑名,dbt 替你管物理名;硬编码就是把自己的 SQL 钉死在某个 schema 上了。

四、source() 深度解析

案例代码

[models/staging/stg_customers.sql]:

-- staging 层: 对 raw_customers 做字段重命名与类型规范,
-- 保持 1:1 投影, 不做业务过滤与聚合.
select
    cast(id as int) as customer_id,
    first_name,
    last_name
from {{ source('raw', 'raw_customers') }}

source() 和 ref() 的区别

	source()	ref()
引用对象	项目外的表(seeds 加载的、外部 ETL 灌入的)	项目内的另一个 model
来源声明	sources.yml 里注册	不需要,ref 的就是项目里的 .sql 文件名
物理名	从 sources.yml 读 schema + table	用 target.schema + custom_schema 算出来

简单记:source() 是"数据从外面进来"的入口,ref() 是"数据在项目里流转"的管道。

source() 干的两件事

1. 声明数据血缘

dbt 在 DAG 里加一条边:raw.raw_customers → stg_customers。这样 dbt docs generate 生成的血缘图能一直追溯到最原始的 seed 表,而不是断在 staging 层。

2. 解析表名

从 [models/staging/sources.yml] 里读配置:

version: 2

sources:
  - name: raw
    # seeds 落在 dbt_dev_raw schema (target.schema=dbt_dev + custom_schema=raw)
    schema: dbt_dev_raw
    description: "源系统原始数据, 由 dbt seed 加载."
    tables:
      - name: raw_customers
        description: "客户主数据原始表."
      - name: raw_orders
        description: "订单原始表."
      - name: raw_payments
        description: "支付流水原始表."

所以 {{ source('raw', 'raw_customers') }} 编译后就是:

from [dbt_dev_raw].[raw_customers]

为什么不直接写表名

把 sources.yml 看成 source() 的"注册表",好处有三个:

1. 血缘可追踪:dbt docs 能画出从 seed → staging → marts 的完整链路。
2. 可做新鲜度测试:source 上可以配 loaded_at_field + freshness,跑 dbt source freshness 检查源表是不是过时了(本项目没配,但机制在那)。
3. 可统一加描述:sources.yml 里的 description 会进 dbt docs,比 SQL 里散落的注释好维护。

五、DAG:依赖图的自动构建

怎么构建的

dbt 启动时会扫描 models/ 下所有 .sql 文件,解析出里面所有的 ref() 和 source() 调用,据此构建一张 DAG(Directed Acyclic Graph,有向无环图):

• 每个 model 是一个节点
• 每次 ref('A') 在 B 里出现 → 一条 A → B 的边
• 每次 source('s','t') 在 B 里出现 → 一条 source 节点 → B 的边

本项目的 DAG

根据上一节读出来的真实 ref() 关系,本项目的依赖图长这样:

   [raw_customers]      [raw_orders]        [raw_payments]        ← sources (Layer 0)
         │                   │                     │
         ▼                   ▼                     ▼
   [stg_customers]     [stg_orders]         [stg_payments]       ← staging (Layer 1)
         │                ╱      ╲             ╱       ╲
         │               ╱        ╲           ╱         ╲
         ▼              ▼          ▼         ▼           ▼
   [dim_customers]                [fct_orders]              ← marts (Layer 2)

把它摊平看依赖关系更清楚:

模型	依赖的 staging	来源 source
stg_customers	—	raw.raw_customers
stg_orders	—	raw.raw_orders
stg_payments	—	raw.raw_payments
dim_customers	stg_customers, stg_orders, stg_payments	—
fct_orders	stg_orders, stg_payments	—

注:第 5 篇里 fct_orders.customer_id 有个 relationships 测试指向 dim_customers,这只是测试依赖(测试要先有 dim_customers 才能跑外键校验),不是模型物化的依赖,所以 fct_orders 本身的 dbt run 不需要 dim_customers 先建好。

DAG 的作用

1. 拓扑排序决定执行顺序:dbt 按拓扑序跑,保证被依赖的模型先建好。本项目执行顺序大致是:raw(seed) → stg_* → dim_customers / fct_orders。
2. 并发跑无依赖的节点:三个 stg_* 之间互不依赖,可以并行(在支持并行的 adapter / 线程配置下)。
3. 改动影响分析:改了 stg_orders,顺着 DAG 一眼看出 dim_customers 和 fct_orders 都受影响,需要重跑——这正是下一节 --select stg_orders+ 做的事。

有环就报错

DAG 的"A"是 Acyclic(无环)。如果 dbt 发现 A 依赖 B、B 又依赖 A(直接或间接),会直接抛 Circular dependency 错误,拒绝跑。这是 dbt 的硬性保护:数据流转不能绕圈,否则就死循环了。

六、--select 选择器:精准控制跑哪些

全量 dbt run 会跑所有模型。项目大了之后你只改了几个模型,没必要每次全跑——这就是 --select(简写 -s)的用武之地。

基本用法

dbt run --select dim_customers

只物化 dim_customers 一个模型。但通常你改了上游,下游也得跟着重建,所以更常用的是带 + 的范围选择。

选择器语法表

语法	含义	示例(本项目)
model_name	只跑这个模型	--select dim_customers
model_name+	这个模型及其所有下游	--select stg_orders+(跑 stg_orders + dim_customers + fct_orders)
+model_name	这个模型及其所有上游	--select +dim_customers(跑 3 个 stg + dim_customers)
+model_name+	上游 + 下游	--select +stg_orders+
path:directory	某目录下所有模型	--select path:models/staging
tag:xxx	按标签筛选	--select tag:nightly
@model_name	上游 + 自己(不含下游)	--select @dim_customers

记忆窍门:+ 在哪边,就往哪边扩。model+ 往下游扩,+model 往上游扩。

实战场景

场景 1:改了 stg_orders,想跑它和所有受影响的下游

dbt run --select stg_orders+

dbt 顺着 DAG 找到 stg_orders 的所有下游(dim_customers、fct_orders),一起重建。

场景 2:staging 已经好了,只想重建 marts 层

dbt run --select path:models/marts

按目录过滤,跑 dim_customers 和 fct_orders,不动 staging。

场景 3:测试某模型及其上游(测试需要上游表先存在)

dbt test --select +dim_customers

+ 表示带上上游,保证测试需要的 stg 表都先建好。注意 dbt test 默认不会自动 run 上游,所以要么先 dbt run --select +dim_customers,要么在已物化的环境里直接 test。

场景 4:组合选择

dbt run --select stg_orders+ dim_customers

多个选择器用空格隔开,取并集。也可以用 intersect、exclude 做交集/差集,进阶用法见官方文档。

七、dbt compile 与 dbt show:看编译结果

Jinja 渲染后的 SQL 长什么样?有两个命令能看。

dbt compile:把模板渲染成纯 SQL

dbt compile --select dim_customers

这不会物化任何表,只把 Jinja 渲染成纯 SQL,写到 target/compiled/ 下。本项目对应文件大致是:

target/compiled/dbt_sqlserver_dw/models/marts/dim_customers.sql

打开后你会看到所有 {{ ref('xxx') }} 都被替换成真实表名:

with customers as (
    select * from [dbt_dev_staging].[stg_customers]
),
orders as (
    select
        customer_id,
        min(order_date) as first_order_date,
        max(order_date) as most_recent_order_date,
        count(order_id) as number_of_orders
    from [dbt_dev_staging].[stg_orders]
    group by customer_id
),
payments as (
    select
        o.customer_id,
        sum(p.amount) as lifetime_value
    from [dbt_dev_staging].[stg_orders] o
    inner join [dbt_dev_staging].[stg_payments] p
        on o.order_id = p.order_id
    where p.status = 'completed'
    group by o.customer_id
)
select ...
from customers c              -- customers/orders/payments 是 CTE 别名,不会被替换
left join orders o on c.customer_id = o.customer_id
left join payments p on c.customer_id = p.customer_id

dbt show:快速预览不建表

dbt show --inline "select * from {{ ref('dim_customers') }} limit 10"

--inline 让你临时塞一段 SQL,dbt 渲染后直接在控制台打印结果,不物化。适合验证一个 ref() 解析得对不对、一个 CTE 出几行。

排查思路

模型行为异常时,标准排查路径:

1. dbt compile --select <模型> 看生成的 SQL 对不对(ref 有没有解析错、schema 对不对)。
2. 把编译后的 SQL 直接拿到 SSMS / Azure Data Studio 里跑一遍,看是 SQL 问题还是 dbt 问题。
3. 如果 SQL 没问题但 dbt run 报错,看是不是物化方式 / 权限 / adapter 的问题。

90% 的"模型不工作"都能在这一步定位。

八、小结

• Jinja 是 dbt 的模板引擎:{{ }} 渲染表达式,{% %} 执行语句;dbt 在其上加了 ref/source/config/var 等几十个函数。
• ref() 干两件事:声明模型间依赖 + 把逻辑名编译成物理表名(用 target.schema + custom_schema 算)。
• source() 干两件事:声明源表→staging 的血缘 + 从 sources.yml 读 schema/table 编译成表名;它还能配 freshness 做新鲜度测试。
• DAG 自动构建:dbt 扫描所有 ref()/source() 调用连边成图,拓扑排序决定执行顺序,有环直接报错。
• --select 选择器:+ 在哪边往哪边扩,path: 按目录,tag: 按标签,@ 只含上游;实战中 model+ 和 +model 用得最多。
• dbt compile / dbt show:看 Jinja 渲染后的真实 SQL,是排查编译和依赖问题的第一手段。