企业官网建设流程全解析

data-engineering-zoomcamp 实战：用 dbt Models 构建星型模型（dim_zones / fct_trips / int_trips_unioned）

【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 👇🏼项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp

本文是>model-paths: ["models"] test-paths: ["tests"] seed-paths: ["seeds"] macro-paths: ["macros"] snapshot-paths: ["snapshots"] models: taxi_rides_ny: staging: +materialized: view intermediate: +materialized: table marts: +materialized: table

其中每个层级都被赋予了不同的物化策略：Staging 用轻量的view，Intermediate 与 Marts 用table——这正是 dbt 分层建模在工程上的落地体现。

先想清楚要构建什么：报表需求与星型模型

在写任何代码之前，先想清楚最终交付物长什么样。课程笔记指出，Marts 层一般承载两类东西：

报表与仪表盘

如果存在一个重要的仪表盘或数据应用——尤其是需要大量手工维护的 Excel/看板——那它就是应该被建模成 dbt Model 的信号。例如“每个地点的月度收入”这类数据集，就应该被建模并纳入版本控制。

维度建模（星型模型）

除了报表，还需要一套规范的星型结构，它包含两类核心表：

表类型	含义	命名前缀	本项目示例
Fact 表	每个事件/过程一行：每趟行程一行、每笔销售一行、每个订单一行	fct_	fct_trips
Dimension 表	某个实体的属性集合	dim_	dim_zones、dim_vendors

星型模型的价值在于：回答“多少个”类问题变得极其简单——COUNT(*)作用于dim_zones即可回答“有多少个 zone”，作用于fct_trips即可回答“有多少趟行程”。单表足够聚焦，复杂查询时再通过 Join 组合。

本项目在 Marts 层最终落地的是：

• dim_zones—— zone/location 属性；
• fct_trips—— 每行一趟行程（yellow + green 合并）；
• 一个报表模型：按 zone 统计月度收入（位于models/marts/reporting/目录）。

source() vs ref()：dbt 依赖图的关键分水岭

这是课程中的关键节点。此前一直使用{{ source() }}拉取原始数据，但source()只适用于在 sources YAML 中声明的、dbt 之外的原始表。如果某个 Model 的输入是另一个 dbt Model，就必须改用{{ ref() }}：

• {{ source('name', 'table') }}→ 读取 YAML 中声明的原始数据；
• {{ ref('model_name') }}→ 读取另一个 dbt Model。

ref()的真正威力在于它的底层副作用：dbt 会基于它自动构建依赖图（dependency graph）。如果模型 B ref 了模型 A，dbt 就知道必须先运行 A 再运行 B——你永远不需要手动维护运行顺序。这在命令上体现为一条链式执行：运行dbt run时，dbt 会先解析所有ref()关系，按拓扑排序依次构建模型。

本项目中的实际依赖链可以完整串联验证这一机制：

source('raw','green_tripdata') / source('raw','yellow_tripdata') │ ref() ▼ stg_green_tripdata / stg_yellow_tripdata │ ref() ▼ int_trips_unioned ──ref()──► int_trips │ ref() ▼ fct_trips ──ref()──► fct_monthly_zone_revenue

可以看到：Staging 模型通过{{ source(...) }}读取原始表；Intermediate 与 Marts 通过{{ ref(...) }}逐层向下游引用，最终构成一张清晰的依赖图。

Intermediate 层：为什么要存在

我们想要fct_trips成为 yellow 与 green 行程数据的并集。但如果把这个 union 直接写进 fact 模型，会让 fact 变得混乱。因此课程把它放进Intermediate（中间层）模型——它既不是 raw，也不直接暴露给终端用户。

• 约定：中间层模型用int_前缀；
• 本例：int_trips_unioned.sql；
• 目的：把中间过程与 Marts 隔离，Marts 只保留“消费就绪”的内容。

课程笔记给出的初始版本如下：

with green_data as ( select *, 'Green' as service_type from {{ ref('stg_green_tripdata') }} ), yellow_data as ( select *, 'Yellow' as service_type from {{ ref('stg_yellow_tripdata') }} ), trips_unioned as ( select * from green_data union all select * from yellow_data ) select * from trips_unioned

注意两点设计细节：

1. 通过'Green'/'Yellow'两个硬编码字符串常量为每一行打上service_type标签，Union 后仍能区分数据来源；
2. 使用union all而非union，因为这里没有去重需求，且能保留所有原始行、避免额外的排序开销。

Union 问题：yellow 与 green 并不完全同构

直接对两个 staging 模型做 union 会报错：set operation can only be applied with expressions with the same number of columns（集合操作只能应用于列数相同的表达式）。原因在于 green 比 yellow 多出两个列：

trip_type

• 取值为1或2；
• 1= 街头招手（street hail）；
• 2= 通过电话或 App 预订（dispatch）；
• yellow 出租车按法规只能街头招手（类型恒为 1），因此原始数据中根本没有这一列；
• 修复：在 yellow 侧补充trip_type并硬编码为1。

ehail_fee（电子叫车附加费）

• 通过 App 叫车时可能产生的附加费用；
• 实践中大部分数据为 NULL——该功能在各服务商间并未统一实现；
• yellow 出租车按定义永远不存在 e-hail 附加费；
• 修复：在 yellow 侧补充ehail_fee并硬编码为0。

课程笔记在 Staging 层的修复方式，是直接改写stg_yellow_tripdata.sql，在 select 列表中补上两列（cast(1 as integer) as trip_type与cast(0 as numeric) as ehail_fee），同时保持与 green staging 一致的类型：

-- Updated stg_yellow_tripdata.sql to match green schema with tripdata as ( select * from {{ source('staging','yellow_tripdata') }} where vendorid is not null ), renamed as ( select -- identifiers cast(vendorid as integer) as vendor_id, cast(ratecodeid as integer) as ratecode_id, cast(pulocationid as integer) as pickup_location_id, cast(dolocationid as integer) as dropoff_location_id, -- timestamps cast(tpep_pickup_datetime as timestamp) as pickup_datetime, cast(tpep_dropoff_datetime as timestamp) as dropoff_datetime, -- trip info store_and_fwd_flag, cast(passenger_count as integer) as passenger_count, cast(trip_distance as numeric) as trip_distance, cast(1 as integer) as trip_type, -- Yellow only does street-hail -- payment info cast(fare_amount as numeric) as fare_amount, cast(extra as numeric) as extra, cast(mta_tax as numeric) as mta_tax, cast(tip_amount as numeric) as tip_amount, cast(tolls_amount as numeric) as tolls_amount, cast(0 as numeric) as ehail_fee, -- Yellow doesn't have ehail cast(improvement_surcharge as numeric) as improvement_surcharge, cast(total_amount as numeric) as total_amount, cast(payment_type as integer) as payment_type, from tripdata ) select * from renamed

修正后的 union 版本：

-- models/staging/int_trips_unioned.sql with green_data as ( select *, 'Green' as service_type from {{ ref('stg_green_tripdata') }} ), yellow_data as ( select *, 'Yellow' as service_type from {{ ref('stg_yellow_tripdata') }} ), trips_unioned as ( select * from green_data union all select * from yellow_data ) select * from trips_unioned

课程笔记特别强调了一个边界：在 Staging 层直接补列，技术上是偏离“1:1 拷贝”原则的。这里是为了保持简单而这样做；在更严格的项目中，列对齐应当在 Intermediate 层完成。有趣的是，仓库的最终实现恰好走的是这条“更严格”的路线（见下文）。

仓库最终实现：列对齐下沉到 Intermediate 层

本仓库实际提交的 stg_yellow_tripdata.sql 并没有补列——它保持与 green 的“差异”不处理（例如只做cast(vendorid as integer)、where vendorid is not null过滤，以及开发环境的时间采样过滤）：

with source as ( select * from {{ source('raw', 'yellow_tripdata') }} ), renamed as ( select cast(vendorid as integer) as vendor_id, cast(ratecodeid as integer) as rate_code_id, cast(pulocationid as integer) as pickup_location_id, cast(dolocationid as integer) as dropoff_location_id, cast(tpep_pickup_datetime as timestamp) as pickup_datetime, cast(tpep_dropoff_datetime as timestamp) as dropoff_datetime, cast(store_and_fwd_flag as string) as store_and_fwd_flag, cast(passenger_count as integer) as passenger_count, cast(trip_distance as numeric) as trip_distance, cast(fare_amount as numeric) as fare_amount, cast(extra as numeric) as extra, cast(mta_tax as numeric) as mta_tax, cast(tip_amount as numeric) as tip_amount, cast(tolls_amount as numeric) as tolls_amount, cast(improvement_surcharge as numeric) as improvement_surcharge, cast(total_amount as numeric) as total_amount, cast(payment_type as integer) as payment_type from source where vendorid is not null ) select * from renamed {% if target.name == 'dev' %} where pickup_datetime >= '2019-01-01' and pickup_datetime < '2019-02-01' {% endif %}

列对齐的工作被真正放到了 int_trips_unioned.sql 中完成。它逐列显式列出两个 CTE 的字段，并在 yellow 分支中硬编码补充差异列：

with green_trips as ( select vendor_id, rate_code_id, pickup_location_id, dropoff_location_id, pickup_datetime, dropoff_datetime, store_and_fwd_flag, passenger_count, trip_distance, trip_type, fare_amount, extra, mta_tax, tip_amount, tolls_amount, ehail_fee, improvement_surcharge, total_amount, payment_type, 'Green' as service_type from {{ ref('stg_green_tripdata') }} ), yellow_trips as ( select vendor_id, rate_code_id, pickup_location_id, dropoff_location_id, pickup_datetime, dropoff_datetime, store_and_fwd_flag, passenger_count, trip_distance, cast(1 as integer) as trip_type, -- Yellow taxis only do street-hail (code 1) fare_amount, extra, mta_tax, tip_amount, tolls_amount, cast(0 as numeric) as ehail_fee, -- Yellow taxis don't have ehail_fee improvement_surcharge, total_amount, payment_type, 'Yellow' as service_type from {{ ref('stg_yellow_tripdata') }} ) select * from green_trips union all select * from yellow_trips

对照可见：

• 课程笔记把补列逻辑放在 Staging（简单但偏离 1:1 原则）；
• 仓库最终实现把补列逻辑放在 Intermediate（符合分层约束，Staging 保持“忠实拷贝 + 类型规范化”）。

两版代码都能解决 union 报错，但后者在工程分层上更规范——这正是一个“文档演示思路、源码给出更严谨落地”的典型对照。

关于类型与 null 值的细节

从仓库源码看，补列时需要注意类型一致性：

• trip_type补cast(1 as integer)——green 侧 stg_green_tripdata.sql 通过{{ safe_cast('trip_type', 'integer') }}将其转为 integer；
• ehail_fee补cast(0 as numeric)——green 侧为cast(ehail_fee as numeric)；
• service_type在两边均为字符串字面量。

此外，sources.yml 的字段描述也印证了业务语义：trip_type（1=Street-hail, 2=Dispatch）、ehail_fee（E-hail fee）仅出现在 green 原始表green_tripdata的列清单中，而 yellow 原始表yellow_tripdata没有这两列。

从 Intermediate 到 Fact：fct_trips 的构建

Union 完成之后，int_trips.sql 负责清洗、富化与去重，为 fact 层提供消费就绪的数据：

with unioned as ( select * from {{ ref('int_trips_unioned') }} ), payment_types as ( select * from {{ ref('payment_type_lookup') }} ), cleaned_and_enriched as ( select {{ dbt_utils.generate_surrogate_key(['u.vendor_id', 'u.pickup_datetime', 'u.pickup_location_id', 'u.service_type']) }} as trip_id, u.vendor_id, u.service_type, u.rate_code_id, u.pickup_location_id, u.dropoff_location_id, u.pickup_datetime, u.dropoff_datetime, u.store_and_fwd_flag, u.passenger_count, u.trip_distance, u.trip_type, u.fare_amount, u.extra, u.mta_tax, u.tip_amount, u.tolls_amount, u.ehail_fee, u.improvement_surcharge, u.total_amount, coalesce(u.payment_type, 0) as payment_type, coalesce(pt.description, 'Unknown') as payment_type_description from unioned u left join payment_types pt on coalesce(u.payment_type, 0) = pt.payment_type ) select * from cleaned_and_enriched qualify row_number() over( partition by vendor_id, pickup_datetime, pickup_location_id, service_type order by dropoff_datetime ) = 1

关键点：

• 用dbt_utils.generate_surrogate_key生成trip_id代理键（依赖 packages.yml 中声明的dbt-labs/dbt_utils）；
• 通过 join seeds 中的payment_type_lookup将付款代码翻译为可读描述；
• 用qualify row_number() = 1做确定性去重。

随后是真正的 Fact 表 fct_trips.sql：

{{ config( materialized='incremental', unique_key='trip_id', incremental_strategy='merge', on_schema_change='append_new_columns' ) }} select trips.trip_id, trips.vendor_id, trips.service_type, trips.rate_code_id, trips.pickup_location_id, pz.borough as pickup_borough, pz.zone as pickup_zone, trips.dropoff_location_id, dz.borough as dropoff_borough, dz.zone as dropoff_zone, trips.pickup_datetime, trips.dropoff_datetime, trips.store_and_fwd_flag, trips.passenger_count, trips.trip_distance, trips.trip_type, {{ get_trip_duration_minutes('trips.pickup_datetime', 'trips.dropoff_datetime') }} as trip_duration_minutes, trips.fare_amount, trips.extra, trips.mta_tax, trips.tip_amount, trips.tolls_amount, trips.ehail_fee, trips.improvement_surcharge, trips.total_amount, trips.payment_type, trips.payment_type_description from {{ ref('int_trips') }} as trips left join {{ ref('dim_zones') }} as pz on trips.pickup_location_id = pz.location_id left join {{ ref('dim_zones') }} as dz on trips.dropoff_location_id = dz.location_id {% if is_incremental() %} where trips.pickup_datetime > (select max(pickup_datetime) from {{ this }}) {% endif %}

这里是星型模型的核心体现：

• Fact + Dimension 的 Join：fct_trips与dim_zones做两次left join（一次取 pickup 地点、一次取 dropoff 地点），把 zone id 富化为 borough/zone 名称。left join保证即使 zone 信息缺失也不会丢 trip 行；
• 增量物化：materialized='incremental'+unique_key='trip_id'+merge策略 +on_schema_change='append_new_columns'，配合is_incremental()条件，只处理 pickup_datetime 晚于当前最大值的增量数据；
• 跨库宏：{{ get_trip_duration_minutes(...) }}封装了 dbt 内置的跨库datediff（见 get_trip_duration_minutes.sql），可在 DuckDB、BigQuery、Snowflake、Redshift、PostgreSQL 等平台无缝运行。

与其配套的 Dimension 表 dim_zones.sql 则保持极简——直接透传 seedtaxi_zone_lookup，但它作为 Model 存在，为将来扩展计算字段、过滤逻辑留了空间：

select locationid as location_id, borough, zone, service_zone from {{ ref('taxi_zone_lookup') }}

报表模型：fct_monthly_zone_revenue

最终面向报表的模型位于 fct_monthly_zone_revenue.sql，它把“每个 zone 的月度收入”固化成了可用 SQL：

select coalesce(pickup_zone, 'Unknown Zone') as pickup_zone, {% if target.type == 'bigquery' %}cast(date_trunc(pickup_datetime, month) as date) {% elif target.type == 'duckdb' %}date_trunc('month', pickup_datetime) {% endif %} as revenue_month, service_type, sum(fare_amount) as revenue_monthly_fare, sum(extra) as revenue_monthly_extra, sum(mta_tax) as revenue_monthly_mta_tax, sum(tip_amount) as revenue_monthly_tip_amount, sum(tolls_amount) as revenue_monthly_tolls_amount, sum(ehail_fee) as revenue_monthly_ehail_fee, sum(improvement_surcharge) as revenue_monthly_improvement_surcharge, sum(total_amount) as revenue_monthly_total_amount, count(trip_id) as total_monthly_trips, avg(passenger_count) as avg_monthly_passenger_count, avg(trip_distance) as avg_monthly_trip_distance from {{ ref('fct_trips') }} group by pickup_zone, revenue_month, service_type

• 使用{{ target.type }}做跨数据库方言的月份截断（BigQuery 与 DuckDB 各自分支），保持了模型的可移植性；
• 用coalesce(pickup_zone, 'Unknown Zone')兜底缺失 zone；
• 按pickup_zone, revenue_month, service_type三维聚合，输出可直接喂给仪表盘的月度收入指标。

业务上下文才是建模决策的依据

yellow 与 green 的列差异绝不只是技术问题，它背后是一段商业故事：纽约出租车牌照制度决定了 yellow cab 主要在曼哈顿运营，green cab 则是为了让外围行政区（outer boroughs）也能打到车而设立。理解了这一点，你才能对trip_type和ehail_fee的处理做出既技术正确又语义正确的决策：

• yellow 按法律只能街头招手 →trip_type恒为1，补列硬编码；
• yellow 按定义不存在 e-hail →ehail_fee恒为0，补列硬编码。

这正是 analytics engineering 区别于普通 SQL 开发的地方：你不再只是写 SQL，而是理解数据到底代表什么。仓库中的 intermediate/schema.yml 与 marts/schema.yml 已经把这种业务理解沉淀为每个字段的描述与数据测试（如service_type的accepted_values: ['Green', 'Yellow']、trip_id的unique/not_null、外键关系的relationships测试等），后续可以配合dbt test持续守护数据质量。

小结：一条可复用的分层建模路径

从本节课程与仓库实现可以提炼出一条通用的 dbt 建模路径：

1. 需求先行：识别报表/仪表盘需求，并规划星型模型（Fact + Dimension）；
2. 分层清晰：Staging（source()读取原始表）→ Intermediate（int_前缀，做 union、清洗、对齐、去重、富化）→ Marts（fct_/dim_前缀，消费就绪）；
3. 依赖交给 dbt：一律用ref()引用上游模型，dbt 自动构建依赖图并排序执行；
4. 列对齐选择正确层级：优先在 Intermediate 层补齐 union 所需的列，保持 Staging 的“忠实拷贝”原则；
5. 物化策略匹配场景：view（staging）/table（intermediate、marts 基础表）/incremental（大表 fact，配合unique_key与merge）；
6. 沉淀业务语义：用 schema.yml 记录字段含义、用 data_tests 固化质量约束、用宏封装跨库逻辑，让模型既正确又可维护。

想要亲手验证以上模型，可以进入 taxi_rides_ny 工程目录，配置 profile 后依次运行dbt run（构建全部模型）与dbt test（执行 schema.yml 中声明的数据测试），观察依赖图如何保证stg_*→int_*→fct_*/dim_*→ 报表模型的正确执行顺序。

【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 👇🏼项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp