企业官网建设流程全解析

dbt 实战入门：基于 Mom's 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 仓库内置的 Mom's Flower Shop 示例项目为核心，系统讲解一个真实 dbt 工程的完整结构：从 dbt_project.yml 配置、CSV seed 数据加载、staging 视图清洗，到分层 analytics 模型（基础聚合与高级业务分析）的实现，再到数据质量测试与静态分析配置。读完本文，你将掌握如何把这份"移植自 SDF 示例"的 dbt 项目跑通，并能照此模式搭建自己的电商/移动应用分析数仓。

项目背景：从 SDF Sample 到 dbt 的移植

Mom's Flower Shop 项目最初是 SDF（SQL 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 流程直接生成这份工程。

数据主题与业务背景

项目包含四类业务数据：

1. Customers（客户）：来自移动 App 的客户信息，如raw_customers.csv（含 id、姓名、邮箱、性别、address_id 等字段，共 1000 条客户记录）；
2. Marketing campaigns（营销活动）：营销活动事件与成本，如raw_marketing_campaign_events.csv（含 campaign_id、campaign_name、c_name（活动类型）、cost 等，共 1480 条）；
3. Mobile in-app events（App 内事件）：用户在 App 内的交互事件，如raw_inapp_events.csv（含 event_id、customer_id、event_time（Unix 毫秒时间戳）、event_name（install / add_to_cart / go_to_checkout / place_order / purchase 等）、event_value、platform、campaign_id，共 16000 条）；
4. 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.sql（stg_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_packages；
• +static_analysis: strict：开启 dbt 的严格静态分析（本次移植相对原 SDF 示例新增的特性），在编译期对模型 SQL 做静态校验，提前发现未引用模型、类型问题等错误；
• staging: +materialized: view + schema: staging：staging 层物化为视图，物理上落到独立stagingschema；
• analytics: +materialized: table：分析层物化为表，落到analyticsschema；
• seeds: +schema: raw：CSV 种子数据落到rawschema，与 staging/analytics 物理隔离。

这种"视图做清洗、表做分析"的物化策略，是 dbt 分层建模的标准实践：staging 层轻量、随源数据实时更新；analytics 层以物化表固化计算，保证下游查询性能。

数据加载：Seeds 与数据质量测试

四个 CSV 位于 seeds/，字段设计如下：

seeds/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(materialized='view') }} 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_type（instagram_ads），并产出活动起止时间、事件数与总花费。

stg_app_installs：安装事件的营销归因

stg_app_installs.sql 将安装事件与营销活动关联，未匹配到活动则标记为自然量（organic）：

SELECT 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 = 'install'

stg_installs_per_campaign：按活动统计安装量

stg_installs_per_campaign.sql 简单按 campaign 分组统计安装数，是后续活动效果分析的基础。

stg_customers：客户 360° 基础视图

stg_customers.sql 将客户与安装信息、地址信息做LEFT JOIN关联，产出客户主数据视图：

{{ config(materialized='view') }} 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均配置unique+not_null，并为主键声明primary_key配置，方便dbt docs生成血缘与数据字典。

基础分析层：analytics_new（7 个模型）

README 将基础分析模型归纳为三类主题：

• 日/时间维度指标：agg_installs_and_campaigns（按日期+活动+平台统计去重安装数）、daily_active_users（DAU/WAU/MAU）、hourly_event_patterns（时段使用规律）；
• 基础表现指标：agg_installs_ranked（活动排名与效果分层）、platform_performance_metrics（平台对比）、geographic_analysis（州级分析）；
• 高管视图：campaign_roi_dashboard（关键指标仪表盘）。

以 agg_installs_and_campaigns.sql 为例（物化为 table）：

{{ config(materialized='table') }} 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,3

campaign_roi_dashboard.sql 则通过三个 CTE（campaign_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( materialized='view', 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 新手入门。

高级分析层：analytics（14 个模型）

高级模型承载复杂业务逻辑，README 按主题归纳为四类 + 特殊特性：

• 客户分析：LTV（customer_lifetime_value）、留存队列（customer_cohort_retention）、RFM 分层（customer_segmentation）、360° 视图（customer_360_view）、转化时长（customer_journey_time）；
• 活动分析：效果汇总（campaign_performance_summary）、对比（campaign_comparison）、ROI、获客成本 CAC（customer_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_score（incremental）；
  • 临时模型：executive_kpi_summary（ephemeral）；
  • 审计表：high_value_customers_audit、daily_revenue_summary_audit（audit_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_rate

macros/calculate_conversion_rate.sql 提供通用转化率计算宏：

{% macro calculate_conversion_rate(numerator, denominator, decimal_places=2) %} 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_*（seeds，rawschema）→stg_*（views，stagingschema）→analytics层（tables，analyticsschema）。dbt build会按依赖顺序自动执行：先加载 seeds、再构建 staging 视图、最后构建 analytics 模型，并同时运行 schema 中定义的所有测试。

数据库配置与快速上手

项目面向"内部分析"场景，README 建议的 profile（internal analytics，形如KW277..）配置如下：

• Database：RAW
• Warehouse：TRANSFORMING
• Schema：moms_flower_shop_<your-name>（每人独立 schema，避免互相污染）
• Role：TRANSFORMER

延迟查询（defer）提示：如需做 compare changes 等演示，可 defer 到生产 schemamoms_flower_shop，复用已构建的上游产物。

运行步骤：

1. 在~/.dbt/profiles.yml中配置 profile（默认名ia_dev），并在dbt_project.yml中确认profile字段与之一致；
2. 将 schema 改为自己的名字（如moms_flower_shop_zhangsan）；
3. 执行构建：

dbt build

dbt build会依次完成：加载 seed → 构建 staging 视图 → 构建 analytics 模型 → 运行数据质量测试。也可按需拆分使用dbt seed、dbt run、dbt test。

常见问题排查

1. Profile not found：确认~/.dbt/profiles.yml中存在ia_dev（或dbt_project.yml中指定的）profile，且凭据完整；
2. 权限错误：确认 TRANSFORMER 角色对 RAW 数据库、TRANSFORMING warehouse 及目标 schema 有相应读写权限；
3. Seed 加载失败：检查 CSV 格式（表头列名需与模型引用一致）、引号转义与编码；
4. 模型编译错误：复查 SQL 语法，确认{{ ref('...') }}引用的模型（含 staging 与 analytics 层）都已存在，必要时检查+static_analysis: strict的静态校验报错信息。

小结：从示例到自建分析工程

Mom's 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