DBT BigQuery 迁移实战：零售数仓管道

如果你在 BigQuery 上用 dbt 构建了数仓管道，迁移到 ClickZetta Lakehouse 的核心工作量集中在 4 处平台差异上，标准 SQL 模型基本不用改。

本文用一个真实迁移项目演示完整过程：将 alanceloth/Retail_Data_Pipeline（Airflow + BigQuery + Cosmos + dbt 的零售数据管道）迁移到 ClickZetta Lakehouse，全部模型经过实际运行验证，e2e 11/11 通过。

完整代码见 GitHub：clickzetta/bigquery2lakehouse-retail

原始项目

alanceloth/Retail_Data_Pipeline 是一个完整的零售数据工程项目，数据来自 Kaggle 的 Online Retail 数据集（英国电商平台 2010 年交易记录，54 万行）。

原始技术栈：

• 编排：Airflow（Astronomer 版）+ Cosmos（自动将 dbt 模型转为 Airflow TaskGroup）
• 存储：Google Cloud Storage（GCS）
• 计算：BigQuery
• 转换：dbt-bigquery
• 数据质量：Soda

数据流：CSV → GCS → BigQuery raw 表 → dbt transform（星型模型）→ dbt report（聚合报表）

dbt 模型共 7 个，构建标准星型模型：

dim_customer

dim_datetime

dim_product

fct_invoices

report_customer_invoices

report_product_invoices

report_year_invoices

迁移后的代码在

03_lakehouse/

01_bigquery/

02_migration/MIGRATION_NOTES.md

结论先行

你的 dbt 项目可以迁移，业务逻辑不需要重写。 这次迁移改动了 5 处，全部是平台配置和函数名替换，7 张模型中 5 张零改动。

profiles.yml

REGEXP_REPLACE

FORMAT_TIMESTAMP

DATE_FORMAT

STRING

varchar

datetime

timestamp

materialized: table

materialized: dynamic_table

整个迁移过程中，编排层的简化比 SQL 改动更显著：原项目需要 Docker + Airflow + Cosmos + GCS + service account JSON，迁移后用

dbt seed

技术栈对比

dbt-bigquery

dbt-clickzetta >= 1.6.5

dbt seed

materialized: table

materialized: dynamic_table

STRING

datetime

TIMESTAMP

varchar

timestamp

FORMAT_TIMESTAMP('%Y-%m-%d', col)

DATE_FORMAT(col, 'yyyy-MM-dd')

EXTRACT(DAYOFWEEK FROM col)

DAYOFWEEK(col)

database: project-id

schema: dataset

schema: schema_name

标准 SQL 操作——SELECT、JOIN、GROUP BY、窗口函数、CTE、

dbt_utils.generate_surrogate_key

准备工作

需要 Python 3.10+ 和 dbt-clickzetta >= 1.6.5。

git clone https://github.com/clickzetta/bigquery2lakehouse-retail.git cd bigquery2lakehouse-retail cp .env.example .env

编辑

.env

.env

CLICKZETTA_SERVICE=cn-shanghai-alicloud.api.clickzetta.com CLICKZETTA_INSTANCE=<your-instance-id> CLICKZETTA_WORKSPACE=<your-workspace> CLICKZETTA_USERNAME=<your-username> CLICKZETTA_PASSWORD=<your-password> CLICKZETTA_SCHEMA=retail CLICKZETTA_VCLUSTER=DEFAULT CZ_PROFILE=retail_dev

一键初始化（创建 cz-cli profile + 生成 dbt profiles.yml）：

cd 03_lakehouse pip install -r requirements.txt python setup.py

验证连接：

cd dbt dbt debug --profiles-dir .

迁移步骤

第一步：数据加载方式替换

原项目的数据加载流程需要 5 个 Airflow Task：

correct_csv_format → upload_retail_csv_to_gcs → create_retail_dataset → retail_gcs_to_raw → country_gcs_to_raw

迁移后用

dbt seed

dbt seed --profiles-dir .

1 of 2 OK loaded seed file retail_raw.country ......... INSERT 239 2 of 2 OK loaded seed file retail_raw.online_retail .... INSERT 14595 Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2

dbt-clickzetta 的 seed 内部走 Volume + COPY INTO，不需要手动配置对象存储、IAM 权限或 service account。

03_lakehouse/dbt/seeds/

dbt_project.yml

+schema: raw

retail_raw

retail

第二步：连接配置替换

BigQuery 的

profiles.yml

project

dataset

retail: outputs: dev: type: bigquery method: service-account keyfile: /path/to/service_account.json project: 'airflow-dbt-soda-pipeline' dataset: retail

ClickZetta 改为 username/password，用

schema

retail: outputs: dev: type: clickzetta service: <your-service-endpoint> instance: <your-instance-id> workspace: <your-workspace> username: <your-username> password: <your-password> schema: retail vcluster: DEFAULT

sources.yml

sources: - name: retail database: 'airflow-dbt-soda-pipeline' # BigQuery：project ID tables: - name: raw_invoices

改为：

sources: - name: retail schema: retail_raw # ClickZetta：schema 名 tables: - name: online_retail # 与 seed 文件名一致 - name: country

第三步：日期格式解析

这是迁移中唯一需要真正思考的地方。

原始 CSV 的

InvoiceDate

12/1/10 8:26

M/D/YY H:MM

TIMESTAMP

TO_TIMESTAMP('12/1/10 8:26', 'M/d/yy H:mm')

原项目的 Airflow DAG 里有一步

correct_csv_format

InvoiceDate

varchar

dim_datetime.sql

REGEXP_REPLACE

dbt_project.yml

seeds: retail: online_retail: +column_types: InvoiceDate: varchar # 先以字符串加载，在模型里转换

dim_datetime.sql

WITH datetime_cte AS ( SELECT DISTINCT InvoiceDate AS datetime_id, TO_TIMESTAMP( REGEXP_REPLACE(InvoiceDate, '(\d+)/(\d+)/(\d+) (\d+):(\d+)', '20$3-$1-$2 $4:$5'), 'yyyy-M-d H:mm' ) AS ts FROM {{ source('retail', 'online_retail') }} WHERE InvoiceDate IS NOT NULL ) SELECT datetime_id, ts AS datetime, DATE_FORMAT(ts, 'dd') AS day, DATE_FORMAT(ts, 'MM') AS month, DATE_FORMAT(ts, 'yyyy') AS year, DATE_FORMAT(ts, 'HH') AS hour, DATE_FORMAT(ts, 'mm') AS minute, DAYOFWEEK(ts) AS weekday FROM datetime_cte

对比 BigQuery 原始写法：

WITH datetime_cte AS ( SELECT DISTINCT CAST(InvoiceDate AS STRING) AS datetime_id, FORMAT_TIMESTAMP('%Y-%m-%d %H:%M:%S', InvoiceDate) AS date_part FROM {{ source('retail', 'raw_invoices') }} WHERE InvoiceDate IS NOT NULL ) SELECT datetime_id, CAST(date_part AS datetime) AS datetime, SUBSTR(date_part, 9, 2) AS day, ... EXTRACT(DAYOFWEEK FROM TIMESTAMP(date_part)) AS weekday FROM datetime_cte

改动点汇总：

CAST(col AS STRING)

FORMAT_TIMESTAMP('%Y-%m-%d %H:%M:%S', col)

REGEXP_REPLACE

TO_TIMESTAMP

CAST(str AS datetime)

TO_TIMESTAMP(...)

SUBSTR(date_part, N, M)

DATE_FORMAT(ts, 'yyyy'/'MM'/'dd')

EXTRACT(DAYOFWEEK FROM TIMESTAMP(col))

DAYOFWEEK(ts)

第四步：Soda → dbt test

原项目在 Airflow DAG 里穿插了 3 次 Soda 数据质量检查（

check_load

check_transform

check_report

迁移后用 dbt 内置的

test

models/schema.yml

models: - name: dim_customer columns: - name: customer_id tests: - unique - not_null - name: fct_invoices columns: - name: customer_id tests: - not_null - relationships: to: ref('dim_customer') field: customer_id

运行：

dbt test --profiles-dir .

Done. PASS=18 WARN=0 ERROR=0 SKIP=0 TOTAL=18

18 个测试覆盖唯一性、非空、引用完整性，替代了原来分散在 Airflow DAG 里的 Soda checks。

编排迁移：Airflow + Cosmos → Studio Tasks

原项目的 Airflow DAG 有 11 个步骤，其中 Cosmos 自动将 dbt 模型依赖转成 TaskGroup：

correct_csv_format → upload_retail_csv_to_gcs → upload_country_csv_to_gcs → create_retail_dataset → retail_gcs_to_raw → country_gcs_to_raw → check_load（Soda）→ transform（Cosmos DbtTaskGroup）→ check_transform（Soda） → report（Cosmos DbtTaskGroup）→ check_report（Soda）

迁移后的架构分两层：

第一层：dbt 负责建表（一次性，或 schema 变更后重建）

dbt seed --profiles-dir . # 加载原始数据 dbt run --profiles-dir . # 创建 7 张动态表

dbt run

CREATE DYNAMIC TABLE ... AS SELECT ...

第二层：Studio Tasks 负责刷新（日常调度）

每个动态表对应一个 Studio Task，内容只有一行：

REFRESH DYNAMIC TABLE workspace.retail.dim_customer;

任务依赖关系镜像 dbt 模型 DAG：

bigquery2lakehouse_retail/ ├── retail_pipeline/ ← 日常刷新（已部署，每日调度） │ ├── 01_dim_customer ─┐ │ ├── 02_dim_datetime ─┼─► 04_fct_invoices ─► 05_report_customer_invoices │ └── 03_dim_product ─┘ ─► 06_report_product_invoices │ ─► 07_report_year_invoices └── retail_pipeline_init/ ← 初始化/重建（草稿，手动执行） ├── init_01_dim_customer （CREATE DYNAMIC TABLE dim_customer AS ...） └── ...

为什么用 REFRESH 而不是 dbt run？

原项目的 Airflow DAG

schedule=None

dbt run

refresh_interval

用

03_lakehouse/tasks/setup.py

cd 03_lakehouse python tasks/setup.py

脚本会：

1. 运行

   dbt compile

   dbt compile
   ，从

   target/run/

   target/run/
   读取 dbt 实际执行的 DDL
2. 生成

   tasks/ddl/

   tasks/ddl/
   （CREATE DYNAMIC TABLE SQL）和

   tasks/refresh/

   tasks/refresh/
   （REFRESH 命令）
3. 在 Studio 创建

   bigquery2lakehouse_retail/retail_pipeline/

   bigquery2lakehouse_retail/retail_pipeline/
   和

   retail_pipeline_init/

   retail_pipeline_init/
   两个目录
4. 设置任务依赖链，deploy 刷新任务

清理所有对象：

python tasks/teardown.py

端到端验证

03_lakehouse/e2e.py

python 03_lakehouse/e2e.py --profile <your-cz-profile>

实际运行结果：

=== e2e verification === [PASS] dim_customer row count: 425 [PASS] dim_datetime row count: 604 [PASS] dim_product row count: 3792 [PASS] fct_invoices row count: 10178 [PASS] top country by revenue: 'United Kingdom' [PASS] top country revenue: 178690.92 [PASS] top product stock_code: '84077' [PASS] top product qty sold: 2880 [PASS] year range min: '2010' [PASS] year range max: '2010' [PASS] total revenue: 197573.37 ======================================== Result: 11/11 checks passed ALL PASSED

11/11 验证通过。

完整验证结果：

dbt seed → Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2 dbt run → Done. PASS=7 WARN=0 ERROR=0 SKIP=0 TOTAL=7 (dynamic_table) dbt test → Done. PASS=18 WARN=0 ERROR=0 SKIP=0 TOTAL=18 e2e → 11/11 checks passed

迁移价值总结

这次迁移不只是"换个数据库"，而是在迁移过程中获得了原项目没有的能力：

基础设施大幅简化

原项目需要维护 Docker + Airflow（Astronomer）+ Cosmos + GCS bucket + IAM 权限 + service account JSON，任何一个环节出问题都会阻塞整个管道。迁移后，数据加载用

dbt seed

从全量重建升级为增量计算

原项目

materialized: table

dbt run

materialized: dynamic_table

编排职责更清晰

原项目 Cosmos 的职责是把 dbt 模型依赖"翻译"成 Airflow Task 依赖，这是一个中间层。迁移后 Studio Tasks 直接持有

REFRESH DYNAMIC TABLE

数据质量内置化

原项目用 Soda 做数据质量检查，需要单独维护 Soda 配置文件和 Python 虚拟环境。迁移后用 dbt test，质量规则和模型定义在同一个项目里，

dbt test

迁移工作量对照

相关文档

• dbt ClickZetta adapter 使用指南
• DBT + ClickZetta Lakehouse 快速入门
• clickzetta/bigquery2lakehouse-retail