Databricks Jobs → Lakehouse Studio 迁移实战：电商 ETL 管道

如果你的数据管道跑在 Databricks Jobs 上——多任务 DAG、任务依赖、定时调度——迁移到 ClickZetta Lakehouse Studio 的核心工作量非常低。任务内容（PySpark/SQL 代码）按 ZettaPark 最小化改写（4 处机械替换），任务编排（DAG 依赖、cron 调度）用

cz-cli

cz-cli

几条命令重建，所有配置一次性完成。

本文用一个真实电商 ETL 管道验证这一点：Bronze 摄取 → Silver 清洗关联 → Gold 聚合，3 个任务 + DAG 依赖 + 每日 02:00 调度，完整迁移到 Lakehouse Studio，经过 8 项自动化验证，全部通过。

完整代码见 GitHub：databricks2lakehouse-jobs

原始项目

01_source/jobs/ecommerce_etl_job.json

01_source/jobs/ecommerce_etl_job.json

：原始 Databricks Jobs 定义，3 个 notebook task，依赖关系 01→02→03，每日凌晨 02:00 触发：

{ "name": "ecommerce_etl_pipeline", "schedule": {"quartz_cron_expression": "0 0 2 * * ?"}, "tasks": [ {"task_key": "ingest_raw", "notebook_task": {...}}, {"task_key": "transform_silver", "depends_on": [{"task_key": "ingest_raw"}]}, {"task_key": "aggregate_gold", "depends_on": [{"task_key": "transform_silver"}]} ], "email_notifications": {"on_failure": ["oncall@company.com"]} }

管道处理电商点击流：500 条事件 × 30 款产品 → 5 个品类的每日销售汇总。

迁移后代码在

03_lakehouse/tasks/

03_lakehouse/tasks/

，可与

01_source/notebooks/

01_source/notebooks/

逐文件对照。

结论先行

改动项	工作量	说明
任务内容（Python 代码）	极低	ZettaPark 4 处替换（import/session/table path/saveAsTable）
任务创建	低	`cz-cli task create --type PYTHON --folder <id>` cz-cli task create --type PYTHON --folder <id>
依赖关系	低	`--dep-tasks '[{"taskId":N,"taskName":"x"}]'` --dep-tasks '[{"taskId":N,"taskName":"x"}]'
Quartz cron → 标准 cron	极低	`"0 0 2 * * ?"` "0 0 2 * * ?" → `"0 2 * * "` "0 2 * *"
告警通知	低	Databricks email → Studio 监控规则（邮件/钉钉/飞书）

dbutils.notebook.run(nb)

dbutils.notebook.run(nb)

、Job cluster 配置——不需要迁移，由 Studio DAG 和 VCluster 自动处理。

技术栈对比

	Databricks Jobs	Lakehouse Studio
管道定义	Job JSON（ `tasks: [...]` tasks: [...] ）	`cz-cli task create/save-config` cz-cli task create/save-config
任务依赖	`depends_on: [{task_key}]` depends_on: [{task_key}]	`--dep-tasks '[{"taskId":N,"taskName":"x"}]'` --dep-tasks '[{"taskId":N,"taskName":"x"}]'
任务内容	Databricks Notebook（PySpark）	Studio Python 任务（ZettaPark）
Session	`spark` spark （全局注入）	`clickzetta_dbutils.get_active_lakehouse_engine()` clickzetta_dbutils.get_active_lakehouse_engine()
调度 cron	Quartz `"0 0 2 * * ?"` "0 0 2 * * ?"	标准 `"0 2 * * "` "0 2 * *"
集群配置	`job_clusters: [{...EC2 配置}]` job_clusters: [{...EC2 配置}]	VCluster 自动管理，无需配置
`dbutils.notebook.run(nb)` dbutils.notebook.run(nb)	串联调用	Studio DAG 依赖替代
失败告警	`email_notifications.on_failure` email_notifications.on_failure	Studio 监控规则（邮件/钉钉/飞书）

迁移步骤

第一步：任务内容 — ZettaPark 4 处改写

每个 notebook 改动极少，业务逻辑完整保留：

# Databricks notebook（原始） from pyspark.sql import functions as F # ← pyspark df = spark.read.csv("/Volumes/ecommerce/landing/events/") # ← spark 全局 events = spark.table("ecommerce.bronze.raw_events") # ← 3 级命名 df.write.saveAsTable("ecommerce.silver.events_enriched")

# Studio Python 任务（03_lakehouse/tasks/02_transform_silver.py） from clickzetta.zettapark import functions as F # ① import # session 由平台注入（通过 clickzetta_dbutils） # ② session events = session.table("jobs_bronze.raw_events") # ③ table path df.write.saveAsTable("jobs_silver.events_enriched") # ④ saveAsTable

DataFrame 逻辑（join/filter/groupBy/agg/withColumn）完全不变。

第二步：任务创建

# Databricks Jobs JSON 中 "task_key" 对应 Studio 任务名 # --type 必填（SQL / PYTHON / SHELL 等） # --folder 指定任务文件夹 ID（通过 cz-cli task list-folders 查询） cz-cli task create etl_01_ingest_raw --type PYTHON --folder 91047 --profile aws_singapore_prod cz-cli task create etl_02_transform_silver --type PYTHON --folder 91047 --profile aws_singapore_prod cz-cli task create etl_03_aggregate_gold --type PYTHON --folder 91047 --profile aws_singapore_prod # 上传任务脚本 cz-cli task save-content <task_id> --file 03_lakehouse/tasks/01_ingest_raw.py

第三步：设置 DAG 依赖

Databricks Jobs 用

depends_on: [{task_key}]

depends_on: [{task_key}]

，Studio 用

--dep-tasks

--dep-tasks

（需要同时提供 taskId 和 taskName）：

# Databricks Job JSON: # {"task_key": "transform_silver", "depends_on": [{"task_key": "ingest_raw"}]} # Studio 等价（需要 taskId + taskName 两个字段）: cz-cli task save-config <id_02> \ --deps replace \ --dep-tasks '[{"taskId":10143594,"taskName":"etl_01_ingest_raw"}]' cz-cli task save-config <id_03> \ --deps replace \ --dep-tasks '[{"taskId":10144488,"taskName":"etl_02_transform_silver"}]'

第四步：调度 cron

Databricks 用 Quartz 6 字段格式，Studio 用标准 5 字段 cron：

# Databricks: "quartz_cron_expression": "0 0 2 * * ?" （秒分时日月周） # Studio: 标准 cron "0 2 * * *" （分时日月周） cz-cli task save-cron <id_01> --cron "0 2 * * *"

第五步：部署

cz-cli task deploy <id_01> # 部署后才能被调度执行 cz-cli task deploy <id_02> cz-cli task deploy <id_03> # 手动触发（等价于 Databricks 的 "Run now"） cz-cli task execute <id_01>

第六步：失败告警配置

Databricks 在 Job JSON 里直接配

email_notifications

email_notifications

，Studio 通过监控规则配置：

Databricks	Studio
Job JSON `email_notifications.on_failure` email_notifications.on_failure	Studio UI：告警监控 → 新建监控规则
只支持邮件	支持邮件、短信、电话（高危）、Webhook（钉钉/飞书）
任务级别配置	可跨任务复用通知策略

配置路径：Studio UI → 运维监控 → 告警监控 → 新建监控规则 → 选择"任务实例失败"事件 → 配置通知方式（邮件/钉钉/飞书 Webhook）

e2e 验证结果

在 AWS 新加坡实例实测，8/8 全部通过：

检查项	期望值	结果
jobs_bronze.raw_events	500	✅
jobs_bronze.products	30	✅
jobs_silver.events_enriched	500	✅
jobs_gold.daily_sales rows	115	✅
总销售额	12,814.84	✅
总订单数	119	✅
品类数	5	✅
Studio 任务全部 ONLINE	3/3	✅

注意事项

--dep-tasks
--dep-tasks
需要 taskId + taskName 两个字段：只传 taskId 会报
```
taskName is required
```
taskName is required
，两者缺一不可。
--folder
--folder
传文件夹 ID 而非名称：通过
```
cz-cli task list-folders
```
cz-cli task list-folders
查询 ID。
任务名不能含斜杠：
```
folder/taskname
```
folder/taskname
格式在 CLI 里会被解析为路径，应用
```
--folder <id>
```
--folder <id>
参数指定文件夹，任务名只写名称部分。
--type
--type
为必填项：
```
cz-cli task create
```
cz-cli task create
不指定
```
--type
```
--type
会报错，常用类型：
```
PYTHON
```
PYTHON
、
```
SQL
```
SQL
、
```
SHELL
```
SHELL
。
cron 格式转换：Quartz 6 字段（秒分时日月周）→ 标准 5 字段（分时日月周）。
```
"0 0 2 * * ?"
```
"0 0 2 * * ?"
→
```
"0 2 * * *"
```
"0 2 * * *"
。