dbt 实战入门基于 Moms Flower Shop 示例项目构建可复用的分析工程dbt-core【免费下载链接】dbtdbt enables data analysts and engineers to transform their data using the same practices that software engineers use to build applications.项目地址: https://gitcode.com/GitHub_Trending/db/dbt本文以 dbt-core 仓库内置的 Moms Flower Shop 示例项目为核心系统讲解一个真实 dbt 工程的完整结构从 dbt_project.yml 配置、CSV seed 数据加载、staging 视图清洗到分层 analytics 模型基础聚合与高级业务分析的实现再到数据质量测试与静态分析配置。读完本文你将掌握如何把这份移植自 SDF 示例的 dbt 项目跑通并能照此模式搭建自己的电商/移动应用分析数仓。项目背景从 SDF Sample 到 dbt 的移植Moms Flower Shop 项目最初是 SDFSQL Data Framework的默认示例工程本仓库将其完整移植到了 dbt 体系见 项目说明。它模拟了一家鲜花电商移动 App 的业务场景数据涵盖客户、营销活动、App 内事件与街道地址四类主题是学习 dbt 分层建模staging → analytics与常见业务指标DAU、留存、LTV、ROAS 等的绝佳教材。在 dbt-core 仓库中该示例位于 crates/dbt-init/assets/moms_flower_shop/并由dbt init的初始化资产机制crates/dbt-init托管意味着用户可通过 dbt 的 init 流程直接生成这份工程。数据主题与业务背景项目包含四类业务数据Customers客户来自移动 App 的客户信息如raw_customers.csv含 id、姓名、邮箱、性别、address_id 等字段共 1000 条客户记录Marketing campaigns营销活动营销活动事件与成本如raw_marketing_campaign_events.csv含 campaign_id、campaign_name、c_name活动类型、cost 等共 1480 条Mobile in-app eventsApp 内事件用户在 App 内的交互事件如raw_inapp_events.csv含 event_id、customer_id、event_timeUnix 毫秒时间戳、event_nameinstall / add_to_cart / go_to_checkout / place_order / purchase 等、event_value、platform、campaign_id共 16000 条Street addresses街道地址客户地址信息如raw_addresses.csv含 address_id、full_address、state、city 等共 500 条。这些原始数据以 CSV 形式存放于 seeds/ 目录通过 dbt seed 机制加载进数仓。工程目录结构全景crates/dbt-init/assets/moms_flower_shop/ ├── dbt_project.yml # dbt 工程配置 ├── macros/ │ └── calculate_conversion_rate.sql # 计算转化率的宏 ├── models/ │ ├── staging/ # 清洗层模型视图 │ │ ├── stg_customers.sql │ │ ├── stg_inapp_events.sql │ │ ├── stg_marketing_campaigns.sql │ │ ├── stg_app_installs.sql │ │ ├── stg_installs_per_campaign.sql │ │ └── schema.yml # 数据质量测试与文档 │ ├── analytics_new/ # 基础分析模型7 个 │ │ ├── agg_installs_and_campaigns.sql │ │ ├── agg_installs_ranked.sql │ │ ├── daily_active_users.sql │ │ ├── platform_performance_metrics.sql │ │ ├── hourly_event_patterns.sql │ │ ├── geographic_analysis.sql │ │ └── campaign_roi_dashboard.sql │ └── analytics/ # 高级分析模型14 个 │ ├── campaign_performance_summary.sql │ ├── campaign_comparison.sql │ ├── customer_acquisition_cost.sql │ ├── customer_lifetime_value.sql │ ├── customer_cohort_retention.sql │ ├── customer_segmentation.sql │ ├── customer_journey_time.sql │ ├── customer_360_view.sql │ ├── event_funnel_analysis.sql │ ├── weekly_growth_metrics.sql │ ├── monthly_revenue_trends.sql │ ├── churn_risk_analysis.sql │ ├── product_affinity_analysis.sql │ ├── user_engagement_score.sql │ ├── session_analysis.sql │ ├── marketing_channel_attribution.sql │ ├── repeat_purchase_analysis.sql │ ├── executive_kpi_summary.sql │ ├── rolling_metrics_snapshot.sql │ ├── high_value_customers_audit.sql │ └── daily_revenue_summary_audit.sql ├── seeds/ # CSV 种子数据 │ ├── raw_customers.csv │ ├── raw_addresses.csv │ ├── raw_inapp_events.csv │ ├── raw_marketing_campaign_events.csv │ └── schema.yml # seed 测试与文档 └── tests/ # 数据质量测试定义于 schema.yml说明README 中的目录树是设计蓝图实际仓库内analytics_new/已并入analytics/目录见 models/analytics/共 17 个 SQL 模型 对应.yml文档文件staging 层另有inapp_events.sqlstg_inapp_events的早期版本。使用时可整体理解为一套 staging → analytics 的两层架构。工程配置dbt_project.yml 逐项解析dbt_project.yml 是工程的配置中枢name: moms_flower_shop version: 1.0.0 config-version: 2 # 工程使用的 profile连接凭据 profile: __PROFILE_NAME__ # 各类文件的查找路径 model-paths: [models] analysis-paths: [analyses] test-paths: [tests] seed-paths: [seeds] macro-paths: [macros] snapshot-paths: [snapshots] target-path: target # 存放编译后 SQL clean-targets: # dbt clean 时删除的目录 - target - dbt_packages models: moms_flower_shop: static_analysis: strict staging: materialized: view schema: staging analytics: materialized: table schema: analytics seeds: moms_flower_shop: schema: raw关键配置解读profile: __PROFILE_NAME__模板占位符dbt init生成工程时会替换为用户实际创建的 profile 名称如ia_dev路径配置声明 models / seeds / macros 等目录target-path保存编译产物dbt clean会清理 target 与 dbt_packagesstatic_analysis: strict开启 dbt 的严格静态分析本次移植相对原 SDF 示例新增的特性在编译期对模型 SQL 做静态校验提前发现未引用模型、类型问题等错误staging: materialized: view schema: stagingstaging 层物化为视图物理上落到独立stagingschemaanalytics: materialized: table分析层物化为表落到analyticsschemaseeds: schema: rawCSV 种子数据落到rawschema与 staging/analytics 物理隔离。这种视图做清洗、表做分析的物化策略是 dbt 分层建模的标准实践staging 层轻量、随源数据实时更新analytics 层以物化表固化计算保证下游查询性能。数据加载Seeds 与数据质量测试四个 CSV 位于 seeds/字段设计如下Seed 文件核心字段行数raw_customers.csvid, first_name, last_name, email, gender, address_id1000raw_addresses.csvaddress_id, full_address, street_number, street_name, state, city500raw_inapp_events.csvevent_id, customer_id, event_time, event_name, event_value, additional_details, platform, campaign_id16000raw_marketing_campaign_events.csvevent_id, event_time, campaign_id, campaign_name, c_name, priority, cost1480seeds/schema.yml 为每个 seed 声明了列级描述与测试seeds: - name: raw_customers description: Raw customer data from the mobile app columns: - name: id description: Unique customer identifier tests: - unique - not_null - name: email tests: - unique - name: raw_addresses columns: - name: address_id tests: - unique - not_null - name: raw_inapp_events columns: - name: event_id tests: - unique - not_null - name: raw_marketing_campaign_events columns: - name: event_id description: Unique marketing event identifier tests: - unique - not_null测试覆盖id/event_id等主键字段的unique、not_null约束email的unique约束。dbt test会把这些断言编译成 SQL 在数据仓库中执行任何重复或空值都会使测试失败。清洗层Staging视图模型逐个拆解staging 层全部物化为视图负责类型转换、字段重命名、表连接与业务口径初加工。stg_inapp_events时间戳清洗stg_inapp_events.sql 把 Unix 毫秒时间戳转换为可读时间{{ config(materializedview) }} SELECT event_id, customer_id, TO_TIMESTAMP(event_time*1000) AS event_time, -- 毫秒 → 时间戳 event_name, event_value, additional_details, platform, campaign_id FROM {{ ref(raw_inapp_events) }}注意原始 CSV 中event_time已是毫秒级整数如1714590000000此处再乘以 1000 为微秒级转换——各数据仓库TO_TIMESTAMP的输入精度不同实际使用时需按目标仓库调整该细节体现了从原 SDF 语法移植时对平台差异的兼容处理。stg_marketing_campaigns活动成本聚合stg_marketing_campaigns.sql 对活动事件做聚合SELECT campaign_id, campaign_name, SUBSTR(c_name, 1, LENGTH(c_name)-1) AS campaign_type, -- 去掉 c_name 末尾下划线 MIN(TO_TIMESTAMP(event_time/1000)) AS start_time, MAX(TO_TIMESTAMP(event_time/1000)) AS end_time, COUNT(event_time) AS campaign_duration, SUM(cost) AS total_campaign_spent, ARRAY_AGG(event_id) AS event_ids FROM {{ ref(raw_marketing_campaign_events) }} GROUP BY campaign_id, campaign_name, campaign_type由于 CSV 中c_name形如instagram_ads_末尾带下划线这里用SUBSTR截掉末字符得到干净的campaign_typeinstagram_ads并产出活动起止时间、事件数与总花费。stg_app_installs安装事件的营销归因stg_app_installs.sql 将安装事件与营销活动关联未匹配到活动则标记为自然量organicSELECT DISTINCT i.event_id, i.customer_id, i.event_time AS install_time, i.platform, COALESCE(m.campaign_id, -1) AS campaign_id, -- 自然量活动记为 -1 COALESCE(m.campaign_name, organic) AS campaign_name, COALESCE(m.c_name, organic) AS campaign_type FROM {{ ref(stg_inapp_events) }} i JOIN {{ ref(raw_marketing_campaign_events) }} m ON (i.campaign_id m.campaign_id) WHERE event_name installstg_installs_per_campaign按活动统计安装量stg_installs_per_campaign.sql 简单按 campaign 分组统计安装数是后续活动效果分析的基础。stg_customers客户 360° 基础视图stg_customers.sql 将客户与安装信息、地址信息做LEFT JOIN关联产出客户主数据视图{{ config(materializedview) }} SELECT c.id AS customer_id, c.first_name, c.last_name, c.first_name || || c.last_name AS full_name, c.email, c.gender, i.campaign_id, i.campaign_name, i.campaign_type, i.platform, -- 营销信息 c.address_id, a.full_address, a.city, a.state -- 地址信息 FROM {{ ref(raw_customers) }} c LEFT OUTER JOIN {{ ref(stg_app_installs) }} i ON (c.id i.customer_id) LEFT OUTER JOIN {{ ref(raw_addresses) }} a ON (c.address_id a.address_id)staging/schema.yml 为这些模型声明了列级文档与测试stg_inapp_events.event_id、stg_marketing_campaigns.campaign_id、stg_customers.customer_id、stg_installs_per_campaign.campaign_id均配置uniquenot_null并为主键声明primary_key配置方便dbt docs生成血缘与数据字典。基础分析层analytics_new7 个模型README 将基础分析模型归纳为三类主题日/时间维度指标agg_installs_and_campaigns按日期活动平台统计去重安装数、daily_active_usersDAU/WAU/MAU、hourly_event_patterns时段使用规律基础表现指标agg_installs_ranked活动排名与效果分层、platform_performance_metrics平台对比、geographic_analysis州级分析高管视图campaign_roi_dashboard关键指标仪表盘。以 agg_installs_and_campaigns.sql 为例物化为 table{{ config(materializedtable) }} SELECT DATE(install_time) AS install_date, campaign_name, platform, COUNT(DISTINCT customer_id) AS distinct_installs FROM {{ ref(stg_app_installs) }} GROUP BY 1,2,3campaign_roi_dashboard.sql 则通过三个 CTEcampaign_summary / campaign_retention / campaign_ltv分别引用campaign_performance_summary、campaign_comparison、customer_acquisition_cost三个高级模型融合 ROAS、7/30 日留存率、LTV、LTV:CAC 比率并用CASE给出活动评级Excellent / Good / Fair / Poor同时打上dashboard、executive标签{{ config( materializedview, tags[dashboard, executive] ) }} ... CASE WHEN cl.ltv_to_cac_ratio 3 AND cr.day_30_retention_rate 40 THEN Excellent WHEN cl.ltv_to_cac_ratio 2 AND cr.day_30_retention_rate 30 THEN Good WHEN cl.ltv_to_cac_ratio 1 AND cr.day_30_retention_rate 20 THEN Fair ELSE Poor END AS campaign_grade这类模型使用基础 CTE 与聚合适合 dbt 新手入门。高级分析层analytics14 个模型高级模型承载复杂业务逻辑README 按主题归纳为四类 特殊特性客户分析LTVcustomer_lifetime_value、留存队列customer_cohort_retention、RFM 分层customer_segmentation、360° 视图customer_360_view、转化时长customer_journey_time活动分析效果汇总campaign_performance_summary、对比campaign_comparison、ROI、获客成本 CACcustomer_acquisition_cost、渠道归因marketing_channel_attribution行为分析事件漏斗event_funnel_analysis、会话分析session_analysis、商品关联product_affinity_analysis、参与度评分user_engagement_score收入分析周/月趋势weekly_growth_metrics/monthly_revenue_trends、复购repeat_purchase_analysis、流失风险churn_risk_analysis特殊特性增量物化user_engagement_scoreincremental临时模型executive_kpi_summaryephemeral审计表high_value_customers_audit、daily_revenue_summary_auditaudit_table。以 campaign_performance_summary.sql物化为 table为例它用三个 CTE 完成活动效果闭环计算campaign_installs从stg_app_installs统计去重安装、首末安装时间排除自然量campaign_id ! -1、campaign_costs从raw_marketing_campaign_events汇总花费与平均成本、post_install_purchases将安装客户与安装后发生的 purchase 事件内连接计算购买人数、总收入与购买次数最终得出每活动 ROI 与转化指标是后续campaign_roi_dashboard的数据源头。复用宏calculate_conversion_ratemacros/calculate_conversion_rate.sql 提供通用转化率计算宏{% macro calculate_conversion_rate(numerator, denominator, decimal_places2) %} ROUND( ({{ numerator }}::FLOAT / NULLIF({{ denominator }}, 0)) * 100, {{ decimal_places }} ) {% endmacro %}使用NULLIF(denominator, 0)避免除零错误除数为 0 时返回 NULL通过::FLOAT强转避免整数除法结果乘以 100 得到百分比decimal_places默认保留 2 位小数可在调用时覆盖在模型中调用{{ calculate_conversion_rate(purchasers, total_installs, 2) }}实现指标口径统一、避免重复 SQL。数据血缘与运行流程整个工程的依赖关系如下Seeds (CSV 文件raw_customers / raw_addresses / raw_inapp_events / raw_marketing_campaign_events) ↓ Staging Models视图stg_ 前缀 ↓ Analytics表 / 视图 / 增量 / 临时模型结合dbt_project.yml模型级血缘为raw_*seedsrawschema→stg_*viewsstagingschema→analytics层tablesanalyticsschema。dbt build会按依赖顺序自动执行先加载 seeds、再构建 staging 视图、最后构建 analytics 模型并同时运行 schema 中定义的所有测试。数据库配置与快速上手项目面向内部分析场景README 建议的 profileinternal analytics形如KW277..配置如下DatabaseRAWWarehouseTRANSFORMINGSchemamoms_flower_shop_your-name每人独立 schema避免互相污染RoleTRANSFORMER延迟查询defer提示如需做 compare changes 等演示可 defer 到生产 schemamoms_flower_shop复用已构建的上游产物。运行步骤在~/.dbt/profiles.yml中配置 profile默认名ia_dev并在dbt_project.yml中确认profile字段与之一致将 schema 改为自己的名字如moms_flower_shop_zhangsan执行构建dbt builddbt build会依次完成加载 seed → 构建 staging 视图 → 构建 analytics 模型 → 运行数据质量测试。也可按需拆分使用dbt seed、dbt run、dbt test。常见问题排查Profile not found确认~/.dbt/profiles.yml中存在ia_dev或dbt_project.yml中指定的profile且凭据完整权限错误确认 TRANSFORMER 角色对 RAW 数据库、TRANSFORMING warehouse 及目标 schema 有相应读写权限Seed 加载失败检查 CSV 格式表头列名需与模型引用一致、引号转义与编码模型编译错误复查 SQL 语法确认{{ ref(...) }}引用的模型含 staging 与 analytics 层都已存在必要时检查static_analysis: strict的静态校验报错信息。小结从示例到自建分析工程Moms Flower Shop 示例完整展示了 dbt 工程的最佳实践闭环seeds 数据落地 → staging 视图清洗含时间戳转换、活动归因、聚合→ analytics 分层建模基础聚合 → 高级业务指标→ schema.yml 数据质量测试与文档 → 宏复用统一口径。配合static_analysis: strict的编译期校验这份工程既适合 dbt 入门练习也适合作为电商/移动应用分析场景的建模蓝本——照此模式你可以把任意业务域的原始 CSV 或表组织成一套可测试、可追溯、可复用的分析流水线。【免费下载链接】dbtdbt enables data analysts and engineers to transform their data using the same practices that software engineers use to build applications.项目地址: https://gitcode.com/GitHub_Trending/db/dbt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考