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 用轻量的viewIntermediate 与 Marts 用table——这正是 dbt 分层建模在工程上的落地体现。先想清楚要构建什么报表需求与星型模型在写任何代码之前先想清楚最终交付物长什么样。课程笔记指出Marts 层一般承载两类东西报表与仪表盘如果存在一个重要的仪表盘或数据应用——尤其是需要大量手工维护的 Excel/看板——那它就是应该被建模成 dbt Model 的信号。例如“每个地点的月度收入”这类数据集就应该被建模并纳入版本控制。维度建模星型模型除了报表还需要一套规范的星型结构它包含两类核心表表类型含义命名前缀本项目示例Fact 表每个事件/过程一行每趟行程一行、每笔销售一行、每个订单一行fct_fct_tripsDimension 表某个实体的属性集合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 了模型 Adbt 就知道必须先运行 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注意两点设计细节通过Green/Yellow两个硬编码字符串常量为每一行打上service_type标签Union 后仍能区分数据来源使用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或21 街头招手street hail2 通过电话或 App 预订dispatchyellow 出租车按法规只能街头招手类型恒为 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 doesnt 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 dont 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) }}将其转为 integerehail_fee补cast(0 as numeric)——green 侧为cast(ehail_fee as numeric)service_type在两边均为字符串字面量。此外sources.yml 的字段描述也印证了业务语义trip_type1Street-hail, 2Dispatch、ehail_feeE-hail fee仅出现在 green 原始表green_tripdata的列清单中而 yellow 原始表yellow_tripdata没有这两列。从 Intermediate 到 Factfct_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( materializedincremental, unique_keytrip_id, incremental_strategymerge, on_schema_changeappend_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 的 Joinfct_trips与dim_zones做两次left join一次取 pickup 地点、一次取 dropoff 地点把 zone id 富化为 borough/zone 名称。left join保证即使 zone 信息缺失也不会丢 trip 行增量物化materializedincrementalunique_keytrip_idmerge策略 on_schema_changeappend_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 的月度收入”固化成了可用 SQLselect 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 建模路径需求先行识别报表/仪表盘需求并规划星型模型Fact Dimension分层清晰Stagingsource()读取原始表→ Intermediateint_前缀做 union、清洗、对齐、去重、富化→ Martsfct_/dim_前缀消费就绪依赖交给 dbt一律用ref()引用上游模型dbt 自动构建依赖图并排序执行列对齐选择正确层级优先在 Intermediate 层补齐 union 所需的列保持 Staging 的“忠实拷贝”原则物化策略匹配场景viewstaging/tableintermediate、marts 基础表/incremental大表 fact配合unique_key与merge沉淀业务语义用 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考