Databricks Notebook → Lakehouse 迁移实战：零售数据 Medallion 管道
如果你的数据工程管道跑在 Databricks Notebook 上，迁移到 ClickZetta Lakehouse Studio 的工作量比你想象的低。Databricks 的 PySpark DataFrame API——
selectselect
、
filterfilter
、
joinjoin
、
withColumnwithColumn
、
whenwhen
、Window 函数——在 ZettaPark 里写法完全一致。改动只有 5 处机械替换：导入路径、Session 获取方式、表路径前缀、一个 API 大小写差异，以及用 Studio 任务依赖替代
dbutils.notebook.run()dbutils.notebook.run()
。
本文用一个真实项目验证这一点：将基于 Databricks 的零售数据 Medallion 管道（Bronze → Silver → Gold 三层架构，14 个 Notebook，81 个代码 Cell）完整迁移到 ClickZetta Lakehouse，提供三种迁移方案，经过 20 项自动化验证，全部通过。
完整代码见 GitHub：databricks2lakehouse-bootcamp
原始项目
databricks2lakehouse-bootcamp fork 自 DataWithBaraa/databricks_bootcamp_2026 （⭐335），原始技术栈是 Databricks + PySpark + Delta Lake + Unity Catalog。项目实现了从 CRM/ERP 双源摄取到星型模型的完整零售数据仓库，覆盖 18,484 名客户、397 款产品、60,398 条销售记录，包含完整的数据清洗、类型转换、代码映射、维度建模。
迁移后代码在
03_lakehouse/03_lakehouse/
目录，可与
01_source/01_source/
逐文件对照。
结论先行
你不需要重写任何业务逻辑，也不需要重新培训团队。5 处改动全是机械替换。
改动项 工作量 说明 导入路径替换 极低 pyspark.sqlpyspark.sql
→ clickzetta.zettaparkclickzetta.zettapark
，全局搜索替换Session 获取方式 极低 sparkspark
（全局注入）→ Studio 任务用 clickzetta_dbutilsclickzetta_dbutils
，本地用 Session.builder.configs({})Session.builder.configs({})
表路径前缀 极低 workspace.bronze.Xworkspace.bronze.X
→ bronze.Xbronze.X
，去掉 catalog 前缀StructField 大小写 极低 field.dataTypefield.dataType
→ field.datatypefield.datatype
（ZettaPark API 差异）编排方式 低 dbutils.notebook.run(nb)dbutils.notebook.run(nb)
→ Studio 任务依赖（DAG）
selectselect
、
filterfilter
、
joinjoin
、
withColumnwithColumn
、
whenwhen
、
coalescecoalesce
、
trimtrim
、
regexp_replaceregexp_replace
、
to_dateto_date
、
castcast
、
isNotNullisNotNull
、Window 函数、
ROW_NUMBER()ROW_NUMBER()
——这些数据工程的核心操作，写法完全一致，不需要改。
技术栈对比
Databricks Notebook Lakehouse Studio 任务 计算引擎 Apache Spark（Databricks） ClickZetta Lakehouse DataFrame API PySpark (pyspark.sqlpyspark.sql
) ZettaPark (clickzetta.zettaparkclickzetta.zettapark
) Session 获取 sparkspark
（Databricks 全局注入）clickzetta_dbutils.get_active_lakehouse_engine()clickzetta_dbutils.get_active_lakehouse_engine()
表命名 workspace.bronze.crm_cust_infoworkspace.bronze.crm_cust_info
（3 级）bronze.crm_cust_infobronze.crm_cust_info
（2 级）文件路径 /Volumes/workspace/bronze/raw_sources/.../Volumes/workspace/bronze/raw_sources/...
vol://bronze.raw_sources/...vol://bronze.raw_sources/...
StructField field.dataTypefield.dataType
field.datatypefield.datatype
SQL 执行 spark.sql(q)spark.sql(q)
立即执行session.sql(q).collect()session.sql(q).collect()
触发执行Notebook 链式调用 dbutils.notebook.run(nb, timeout_seconds=0)dbutils.notebook.run(nb, timeout_seconds=0)
Studio 任务依赖（--deps--deps
参数） 调度编排 Databricks Jobs Studio 任务 DAG
项目背景
数据来自自行车零售商 CRM + ERP 双源系统，包含 6 个 CSV 文件：
数据源 表名 行数 说明 CRM crm_cust_infocrm_cust_info
18,494 客户信息（含脏数据和大小写不一致） CRM crm_prd_infocrm_prd_info
397 产品信息（含编码拼接的品类 ID） CRM crm_sales_detailscrm_sales_details
60,398 销售流水（含 yyyyMMdd 格式日期） ERP erp_cust_az12erp_cust_az12
18,484 客户主数据（含 NAS 前缀、未来日期） ERP erp_loc_a101erp_loc_a101
18,484 客户地址（含连字符和国家代码缩写） ERP erp_px_cat_g1v2erp_px_cat_g1v2
37 产品品类（含 YES/NO 维护标志）
Medallion 架构分三层：
Bronze ：原始 CSV → 6 张 Delta 表（无转换）Silver ：清洗 + 标准化 → 6 张宽表（trim、类型转换、枚举映射、列重命名）Gold ：星型模型 → dim_customers + dim_products + fact_sales
原始项目有 14 个 Databricks Notebook（81 个代码 Cell），通过
dbutils.notebook.run()dbutils.notebook.run()
串联执行。
迁移步骤
第一步：替换导入路径
机械全局替换，无逻辑改动：
# Databricks
import pyspark.sql.functions as F
from pyspark.sql.types import StringType, DateType
from pyspark.sql.functions import trim, col, length
from pyspark.sql.window import Window
# ZettaPark（仅包名不同）
from clickzetta.zettapark import functions as F
from clickzetta.zettapark.types import StringType, DateType
from clickzetta.zettapark.functions import trim, col, length
from clickzetta.zettapark.window import Window
第二步：替换 Session 获取方式
Databricks 把
sparkspark
注入到每个 Notebook，不需要显式创建。Studio 任务通过
clickzetta_dbutilsclickzetta_dbutils
获取，同样不需要管密码或连接参数：
# Databricks：spark 全局可用，直接用
df = spark.table("workspace.bronze.crm_cust_info")
# Studio 任务：平台注入，通过 clickzetta_dbutils 获取 session
from clickzetta_dbutils import get_active_lakehouse_engine
from clickzetta.zettapark.session import Session
from urllib.parse import urlparse, parse_qs
engine = get_active_lakehouse_engine(schema="quick_start")
url_str = str(engine.url)
parsed = urlparse(url_str.replace('clickzetta://', 'https://'))
params = parse_qs(parsed.query)
parts = parsed.hostname.split('.', 1)
session = Session.builder.configs({
"service": parts[1],
"instance": parts[0],
"magic_token": params['magic_token'][0],
"workspace": parsed.path.lstrip('/'),
"schema": params.get('schema', ['quick_start'])[0],
"vcluster": params.get('virtualcluster', ['DEFAULT'])[0],
}).getOrCreate()
# 之后 session 的用法与 spark 完全一致
df = session.table("bronze.crm_cust_info")
💡 提示 ：本地开发调试时，用
Session.builder.configs({"instance":..., "password":..., ...}).create()Session.builder.configs({"instance":..., "password":..., ...}).create()
显式传入凭据，不依赖
clickzetta_dbutilsclickzetta_dbutils
。两种获取方式之后的 DataFrame 代码完全相同。
第三步：去掉表路径的 catalog 前缀
Unity Catalog 用三级命名（catalog.schema.table），Lakehouse 在单 workspace 下只需两级：
# Databricks（Unity Catalog 三级）
df = spark.table("workspace.bronze.crm_cust_info")
df.write.mode("overwrite").format("delta").saveAsTable("workspace.silver.crm_customers")
# ZettaPark（两级，去掉 catalog 前缀）
df = session.table("bronze.crm_cust_info")
df.write.mode("overwrite").saveAsTable("silver.crm_customers") # .format("delta") 也不需要
第四步：修正 StructField 大小写
这是实测发现的一处 API 差异，影响所有用
df.schema.fieldsdf.schema.fields
迭代字段类型的代码：
# Databricks / PySpark
for field in df.schema.fields:
if isinstance(field.dataType, StringType): # 大写 T
df = df.withColumn(field.name, trim(col(field.name)))
# ZettaPark（datatype 全小写）
for field in df.schema.fields:
if isinstance(field.datatype, StringType): # 小写 t
df = df.withColumn(field.name, trim(col(field.name)))
第五步：用 Studio 任务依赖替代 dbutils.notebook.run()
原始项目用
dbutils.notebook.run()dbutils.notebook.run()
串联 14 个 Notebook，直接对应 Studio 任务 DAG：
# Databricks：orchestration notebook
notebooks = [
"./silver_crm_cust_info",
"./silver_crm_prd_info",
"./silver_crm_sales_details",
"./silver_erp_cust_az12",
"./silver_erp_loc_a101",
"./silver_erp_px_cat_g1v2"
]
for nb in notebooks:
dbutils.notebook.run(nb, timeout_seconds=0)
# Studio：用 cz-cli 设置任务依赖，一次性配置，之后由平台调度
cz-cli task save-config bootcamp/silver_crm_cust_info \
--deps bootcamp/bronze_ingestion --profile aws_singapore_prod
cz-cli task save-config bootcamp/gold_dim_customers \
--deps bootcamp/silver_crm_cust_info,bootcamp/silver_erp_cust_az12,bootcamp/silver_erp_loc_a101 \
--profile aws_singapore_prod
执行 DAG（等价于原 orchestration notebook 的串联调用）：
cz-cli task execute bootcamp/init_lakehouse --profile aws_singapore_prod
# → 自动触发 bronze → silver（并行）→ gold（按依赖顺序）
迁移后的 Studio 任务 DAG
原始 Databricks 用 2 个 orchestration notebook 串联，迁移后 Studio 平台自动管理依赖和并行度：
init_lakehouse (SQL 任务)
└── bronze_ingestion (Python 任务)
├── silver_crm_cust_info ← 并行执行
├── silver_crm_prd_info ← 并行执行
├── silver_crm_sales_details ← 并行执行
├── silver_erp_cust_az12 ← 并行执行
├── silver_erp_loc_a101 ← 并行执行
└── silver_erp_px_cat_g1v2 ← 并行执行
├── gold_dim_customers ← 依赖 silver_crm_cust_info + erp_cust + erp_loc
├── gold_dim_products ← 依赖 silver_crm_prd_info + erp_px_cat
└── gold_fact_sales ← 依赖 dim_customers + dim_products + silver_crm_sales
原始 Databricks 串行执行 6 个 silver notebook，Studio 并行执行 6 个 silver 任务——相同逻辑，更短的整体运行时间。
完全兼容的部分
以下代码在两边写法完全一致，实测验证通过，不需要改：
# 字符串清洗
for field in df.schema.fields:
if isinstance(field.datatype, StringType):
df = df.withColumn(field.name, trim(col(field.name)))
# 枚举映射（条件替换）
df = df.withColumn("cst_marital_status",
F.when(F.upper(F.col("cst_marital_status")) == "S", "Single")
.when(F.upper(F.col("cst_marital_status")) == "M", "Married")
.otherwise("n/a"))
# 复合字符串解析（品类 ID 从产品编码中提取）
df = df.withColumn("cat_id", F.regexp_replace(F.substring(col("prd_key"), 1, 5), "-", "_"))
df = df.withColumn("prd_key", F.substring(col("prd_key"), 7, F.length(col("prd_key"))))
# 日期格式转换（yyyyMMdd 整数 → DATE）
df = df.withColumn("sls_order_dt",
F.when(
(col("sls_order_dt") == 0) | (length(col("sls_order_dt")) != 8), None
).otherwise(F.to_date(col("sls_order_dt").cast("string"), "yyyyMMdd")))
# 条件价格修复（quantity != 0 时从 sales/quantity 推算）
df = df.withColumn("sls_price",
F.when(
(col("sls_price").isNull()) | (col("sls_price") <= 0),
F.when(col("sls_quantity") != 0, col("sls_sales") / col("sls_quantity")).otherwise(None)
).otherwise(col("sls_price")))
# 前缀清洗（去掉 NAS 开头的无效前缀）
df = df.withColumn("cid",
F.when(col("cid").startswith("NAS"), F.substring(col("cid"), 4, F.length(col("cid"))))
.otherwise(col("cid")))
# 未来日期过滤（脏数据清洗）
df = df.withColumn("bdate",
F.when(col("bdate") > F.current_date(), None).otherwise(col("bdate")))
# 多表 LEFT JOIN + ROW_NUMBER 建维度表
df = session.sql("""
SELECT
ROW_NUMBER() OVER (ORDER BY ci.customer_id) AS customer_key,
ci.customer_id, ci.customer_number, ci.first_name, ci.last_name,
COALESCE(la.country, 'n/a') AS country, ci.marital_status,
CASE WHEN ci.gender <> 'n/a' THEN ci.gender ELSE COALESCE(ca.gender, 'n/a') END AS gender,
ca.birth_date, ci.created_date
FROM silver.crm_customers ci
LEFT JOIN silver.erp_customer_location la ON ci.customer_number = la.customer_number
LEFT JOIN silver.erp_customers ca ON ci.customer_number = ca.customer_number
""")
e2e 验证结果
在 AWS 新加坡实例（
aws_singapore_prodaws_singapore_prod
）实测，20/20 全部通过：
检查项 期望值 结果 bronze.crm_cust_info 18,494 ✅ bronze.crm_prd_info 397 ✅ bronze.crm_sales_details 60,398 ✅ bronze.erp_cust_az12 18,484 ✅ bronze.erp_loc_a101 18,484 ✅ bronze.erp_px_cat_g1v2 37 ✅ silver.crm_customers 18,490 ✅ silver.crm_products 397 ✅ silver.crm_sales 60,398 ✅ silver.erp_customers 18,484 ✅ silver.erp_customer_location 18,484 ✅ silver.erp_product_category 37 ✅ gold.dim_customers 18,490 ✅ gold.dim_products 397 ✅ gold.fact_sales 89,833 ✅ 总销售额 43,538,800 ✅ 负销售额行数 5（原始数据） ✅ 不重复客户数 18,484 ✅ 不重复产品 SKU 295 ✅ 空订单日期行数 24（原始数据格式问题） ✅
Bronze 18,494 → Silver 18,490：清洗时过滤了 4 条 NULL customer_id 脏数据，与原始 Databricks 行为一致。
三种迁移方案对比
本项目提供三种方案，适合不同团队的技术背景和部署需求：
方案 文件位置 Session 方式 改动量 适用场景 A. ZettaPark 本地 03_lakehouse/{bronze,silver,gold}/03_lakehouse/{bronze,silver,gold}/
Session.builder.configs({}).create()Session.builder.configs({}).create()
~5% 本地开发、CI/CD 调试 B. 纯 SQL 03_lakehouse/sql/03_lakehouse/sql/
不需要 全改写（逻辑不变） SQL 优先团队、Studio SQL 任务 C. Studio 任务 03_lakehouse/tasks/03_lakehouse/tasks/
clickzetta_dbutilsclickzetta_dbutils
~5% 生产部署（推荐）
三种方案产生完全相同的结果——行数、指标、数据一致。方案 C 是生产环境的推荐路径，与 Databricks Notebook + Jobs 的架构直接对应。
注意事项
field.datatypefield.datatype
大小写StructFieldStructField
属性是 datatypedatatype
（全小写），PySpark 是 dataTypedataType
（驼峰）。这是唯一一处非机械替换的 API 差异，需要手动搜索 schema.fieldsschema.fields
的用法逐一修改。Studio 任务的 clickzetta_dbutilsclickzetta_dbutils
模板 ：get_active_lakehouse_engine()get_active_lakehouse_engine()
调用返回当前任务的连接信息，需要解析 URL 构建 Session。这段模板代码固定，可以提取到公共模块复用。Volume 路径格式 ：Databricks /Volumes/catalog/schema/volume/path/Volumes/catalog/schema/volume/path
→ ZettaPark vol://schema.volume/pathvol://schema.volume/path
，注意去掉 catalog 层级。.format("delta").format("delta")
不需要saveAsTable()saveAsTable()
默认写 Lakehouse 原生格式，不需要显式指定。Silver 行数差异（18,494 → 18,490） ：CRM 源数据中有 4 条 NULL customer_id 记录，Silver 层清洗时过滤，属于预期行为。
相关文档
ZettaPark DataFrame API
Studio 任务与编排
其他迁移实战
