dbt-databricks → dbt-clickzetta 迁移实战：金融支付数据管道
如果你的数据管道用 dbt + Databricks SQL 构建，迁移到 ClickZetta Lakehouse 的工作量比你想象的低——低到只改
profiles.yml
profiles.yml
里的一行。dbt 的所有核心能力——CTE 模型、窗口函数、SCD Type 2 逻辑、数据质量测试、单元测试、数据契约——在 Lakehouse 上完全兼容，一字不改。
本文用一个真实项目验证这一点：将基于 dbt-databricks 的金融支付数据管道（staging → intermediate → marts → semantic，30 个模型，36 项测试）完整迁移到 dbt-clickzetta，经过
dbt seed 9/9 + dbt run 30/30 + dbt test 36/36
dbt seed 9/9 + dbt run 30/30 + dbt test 36/36
验证，全部通过。
完整代码见 GitHub：dbt-databricks2lakehouse-blueprint
原始项目
dbt-databricks2lakehouse-blueprint fork 自 Alex-Teodosiu/dbt-blueprint（⭐72），原始技术栈是 dbt-databricks + Databricks SQL Warehouse。项目模拟了一个 P2P/P2B（个人转账/企业付款）支付平台的数据工程管道，覆盖用户、商户、账户、银行卡和交易 5 个数据域，实现了从原始事件 → 清洗 → SCD Type 2 维度表 → 事实表 → 语义层的完整链路。
迁移后的代码在
03_lakehouse/
03_lakehouse/
目录，可与
01_source/dbt_blueprint/
01_source/dbt_blueprint/
逐文件对照。
结论先行
adapter 是 dbt 项目的数据库方言适配层——换 adapter 就是换数据库，dbt 模型本身不需要改。改动集中在
profiles.yml
profiles.yml
（连接配置）和少量 Databricks 特有语法。
| 改动项 | 文件数 | 工作量 | 说明 |
|---|
| adapter 切换 | 1（profiles.yml
profiles.yml ） | 极低 | type: databricks
type: databricks → type: clickzetta
type: clickzetta ，连接参数不同 |
| 项目 profile 名 | 1（dbt_project.yml
dbt_project.yml ） | 极低 | connection_databricks
connection_databricks → dbt_blueprint
dbt_blueprint |
getdate()
getdate() | 1 | 极低 | → current_date()
current_date() |
| 宏目标名 | 1 | 极低 | databricks_cluster
databricks_cluster → clickzetta_prod
clickzetta_prod |
完全不需要改的部分：所有 CTE 模型逻辑、窗口函数（LEAD/LAG/ROW_NUMBER）、SCD Type 2 逻辑、
dbt_utils.generate_surrogate_key()
dbt_utils.generate_surrogate_key()
、数据质量测试、单元测试、数据契约、
SELECT * EXCEPT (col)
SELECT * EXCEPT (col)
、
col :: type
col :: type
cast 语法、
DATEDIFF(year, start, end)
DATEDIFF(year, start, end)
——Lakehouse 全部原生支持。
技术栈对比
| dbt-databricks（原始） | dbt-clickzetta（迁移后） |
|---|
| dbt adapter | dbt-databricks
dbt-databricks | dbt-clickzetta
dbt-clickzetta |
| 计算引擎 | Databricks SQL Warehouse | ClickZetta Lakehouse |
| 连接方式 | host
host + http_path
http_path + token
token | instance
instance + workspace
workspace + service
service + username
username + password
password |
| 目标 catalog | catalog: dbt_blueprint
catalog: dbt_blueprint | 不需要（workspace 即 catalog） |
| SQL 方言 | Databricks SQL | Lakehouse SQL（ANSI 兼容） |
:: cast
:: cast | 支持 | 支持（两边一致，无需改） |
SELECT * EXCEPT
SELECT * EXCEPT | 支持 | 支持（两边一致，无需改） |
DATEDIFF(year, s, e)
DATEDIFF(year, s, e) | 支持 | 支持（三参数形式兼容） |
getdate()
getdate() | 支持 | 不支持 → current_date()
current_date() |
| 模型逻辑（CTE/Window/JOIN） | — | 完全一致 |
项目背景
数据来自模拟的 P2P/P2B 支付平台，包含 9 张原始事件表：
| 数据域 | 表 | 行数 | 说明 |
|---|
| 用户 | user_events
user_events | 196 | 用户注册/状态变更事件（SCD2 源） |
| 商户 | merchant_events
merchant_events | 30 | 商户注册/状态变更 |
| 商户 | industry_codes
industry_codes | 30 | 行业代码参考表 |
| 账户 | account_events
account_events | 80 | 银行账户开户/变更事件 |
| 银行卡 | card_events
card_events | 60 | 银行卡开通/状态变更 |
| 交易 | raw_p2p_captured_events
raw_p2p_captured_events | 150 | P2P 成功交易 |
| 交易 | raw_p2p_failed_events
raw_p2p_failed_events | 50 | P2P 失败交易 |
| 交易 | raw_p2b_captured_events
raw_p2b_captured_events | 80 | P2B 成功交易 |
| 交易 | raw_p2b_failed_events
raw_p2b_failed_events | 30 | P2B 失败交易 |
dbt 四层架构：
staging → intermediate → marts → semantic
(view) (table) (table) (table)
raw events SCD2 logic dim + fact enriched
迁移步骤
第一步：切换 adapter（核心改动）
原始
profiles.yml
profiles.yml
用
type: databricks
type: databricks
，迁移只需换成
type: clickzetta
type: clickzetta
，连接参数从 Databricks 的
host/http_path/token
host/http_path/token
换成 Lakehouse 的
instance/workspace/service
instance/workspace/service
：
# 原始（dbt-databricks）
connection_databricks:
target: dev_local
outputs:
dev_local:
type: databricks
catalog: dbt_blueprint
schema: default
host: dbc-a505ff10-0af5.cloud.databricks.com
http_path: /sql/1.0/warehouses/4fa4ca06332da87f
token: "{{ env_var('BLUEPRINT_DATABRICKS_TOKEN') }}"
threads: 1
# 迁移后（dbt-clickzetta）
dbt_blueprint:
target: dev
outputs:
dev:
type: clickzetta
instance: "{{ env_var('CZ_INSTANCE') }}"
workspace: "{{ env_var('CZ_WORKSPACE') }}"
schema: "{{ env_var('CZ_SCHEMA', 'dbt_blueprint_dev') }}"
vcluster: "{{ env_var('CZ_VCLUSTER', 'DEFAULT') }}"
username: "{{ env_var('CZ_USERNAME') }}"
password: "{{ env_var('CZ_PASSWORD') }}"
service: "{{ env_var('CZ_SERVICE') }}"
threads: 4
同步更新
dbt_project.yml
dbt_project.yml
里的
profile
profile
名称：
# 原始
profile: 'connection_databricks'
# 迁移后
profile: 'dbt_blueprint'
第二步：替换 getdate()
getdate()
getdate()
getdate()
是 SQL Server/Databricks 的函数，Lakehouse 用
current_date()
current_date()
：
-- 原始（int_users.sql）
{{ calculate_age('date_of_birth', 'getdate()') }} as age
-- 迁移后
{{ calculate_age('date_of_birth', 'current_date()') }} as age
💡 提示：
:: cast
:: cast
语法和
DATEDIFF(year, start, end)
DATEDIFF(year, start, end)
三参数形式在 Lakehouse 均
原生支持，不需要改。
第三步：更新 generate_schema_name
generate_schema_name
宏中的目标名
原始宏里 prod 环境名是
databricks_cluster
databricks_cluster
，改为
clickzetta_prod
clickzetta_prod
：
-- 原始
{%- if target.name in ["prod", "databricks_cluster"] -%}
-- 迁移后
{%- if target.name in ["prod", "clickzetta_prod"] -%}
完全兼容的部分（不需要改）
以下模式在两边写法完全一致，全部实测通过：
-- :: cast 类型转换 — 两边均支持
userId :: string as user_id,
eventTime :: timestamp as event_time,
amount :: double as transaction_amount
-- SCD Type 2 窗口逻辑
, users_scd2 as (
select
user_id, status, event_time as from_event_timestamp,
lead(event_time) over w as to_event_timestamp
from users
window w as (partition by user_id order by event_time)
)
-- DATEDIFF 三参数 — 两边均支持
DATEDIFF(year, date_of_birth, current_date())
-- Surrogate key（dbt_utils）
{{ dbt_utils.generate_surrogate_key(['user_id', 'from_event_timestamp', 'status']) }}
-- SELECT * EXCEPT — 两边均支持
select * except (age)
from users
where age >= 18
-- 数据契约（model contracts）
config:
contract:
enforced: true
columns:
- name: transaction_uid
data_type: string
data_tests:
- not_null
- unique
-- 单元测试（unit tests）
unit_tests:
- name: test__int_users__scd_logic
model: int_users
given:
- input: ref('stg_user_events')
rows: [...]
expect:
rows: [...]
dbt test 验证结果
在 AWS 新加坡实例（
aws_singapore_prod
aws_singapore_prod
）实测，36/36 全部通过：
dbt seed: 9/9 PASS (196 users, 310 transactions, 80 accounts, 60 cards...)
dbt run: 30/30 PASS (9 views + 21 tables)
dbt test: 36/36 PASS (21 data tests + 15 unit tests)
数据质量测试包括：源表 not_null/unique 约束、marts 层数据契约（
contract: enforced: true
contract: enforced: true
）、fact_transaction 列类型验证。单元测试覆盖了 SCD2 逻辑（
int_users__scd_logic
int_users__scd_logic
）、交易聚合（
int_transactions__union_all
int_transactions__union_all
）、年龄计算宏（
calculate_age__valid_ages
calculate_age__valid_ages
）。
注意事项
getdate()
getdate()
vs current_date()
current_date()
：Databricks 的 getdate()
getdate()
返回当前时间戳，Lakehouse 用 current_date()
current_date()
（日期）或 current_timestamp()
current_timestamp()
（时间戳）替代。col :: type
col :: type
cast 语法和 DATEDIFF(year, s, e)
DATEDIFF(year, s, e)
三参数形式在 Lakehouse 均原生支持，不需要改。- catalog 层级：Databricks 用三级命名（
catalog.schema.table
catalog.schema.table
），Lakehouse 在 dbt 中只用 schema.table
schema.table
，adapter 会自动处理。无需手动修改模型里的表引用。
- seeds 替代 ingestion 层：原始项目依赖 Databricks workspace 的摄取表。迁移到 Lakehouse 后，用 seeds CSV 替代，生产环境可换成 Studio 数据集成或 CDC 任务。
相关文档
dbt 开发指南
其他迁移实战
