Databricks DLT → Lakehouse 迁移实战：服装零售流式管道

如果你的数据管道跑在 Databricks Delta Live Tables 上，迁移到 ClickZetta Lakehouse 的核心工作量比你想象的低——大部分 PySpark DataFrame 代码可以直接复用。DLT 用装饰器（

@dlt.table

@dlt.expect_or_drop

df.write.saveAsTable()

本文用一个真实项目验证这一点：将基于 Databricks DLT 的服装零售流式管道（Bronze 摄取 → Silver SCD2 清洗 → Gold 聚合分析）完整迁移到 ClickZetta Lakehouse，经过 16 项自动化验证，全部通过。

完整代码见 GitHub：databricks2lakehouse-dlt-apparel

原始项目

databricks2lakehouse-dlt-apparel fork 自 jrlasak/databricks_apparel_streaming（⭐45），原始技术栈是 Databricks DLT + Auto Loader + Delta Lake。项目实现了服装零售商的完整数据管道，包含 4 个维度（客户、商品、门店、交易），覆盖流式摄取、SCD Type 2 历史追踪、数据质量约束、Gold 层聚合分析。

迁移后的代码在

03_lakehouse/

01_source/dlt/

结论先行

方案 C（Dynamic Table）是 DLT 管道最接近的对标——声明式定义 + 自动定时刷新，跟 DLT 的"定义即执行"理念一致。如果现有团队习惯 PySpark DataFrame，方案 A（ZettaPark）只需 5 处机械替换，业务逻辑完整保留。

03_lakehouse/*.py

@dlt.table

df.write.saveAsTable()

03_lakehouse/sql/

03_lakehouse/dynamic_tables/

方案 A 改动清单（ZettaPark 主线）：

import dlt

from pyspark...

from clickzetta.zettapark import functions as F

spark

session = Session.builder.configs({}).create()

@dlt.table(name=X)

df.write.mode("overwrite").saveAsTable(X)

@dlt.expect_or_drop("msg","cond")

df.filter(F.col(...)...)

dlt.read_stream("LIVE.X")

dlt.read("LIVE.X")

session.table("X")

dlt.create_auto_cdc_flow(scd_type=2)

F.lead().over(Window.partitionBy(key).orderBy(seq))

F.window("event_time","1 day")

F.to_date(F.col("event_time"))

session.read.csv("vol://...")

JOIN 逻辑、聚合函数（

F.sum/count/avg

F.when/coalesce/lead/lag

Window

技术栈对比

@dlt.table

df.write.mode("overwrite").saveAsTable(X)

@dlt.expect_or_drop("msg","cond")

df.filter(condition)

dlt.read_stream("LIVE.X")

session.table("X")

dlt.read("LIVE.X")

session.table("X")

dlt.create_auto_cdc_flow(stored_as_scd_type=2)

F.lead().over(Window.partitionBy(key).orderBy(seq))

F.window("event_time","1 day")

F.to_date(F.col("event_time"))

cloudFiles

session.read.csv("vol://...")

预览

项目背景

服装零售商，4 个数据流 + Bronze/Silver/Gold 三层架构：

raw_customers

raw_products

raw_stores

raw_sales

DLT 管道文件结构：

01_bronze.py — Auto Loader 流摄取 → 4 张 Bronze 表 02A_silver.py — @dlt.view + @dlt.expect_or_drop（数据质量过滤） 02B_silver.py — dlt.create_auto_cdc_flow（SCD Type 2 维度表） 02C_silver.py — 销售事务清洗 03_gold.py — 4 张 Gold 聚合表

迁移步骤

第一步：

@dlt.table

@dlt.table

/ Auto Loader → ZettaPark

DLT 用装饰器声明表 +

spark.readStream

session.read.csv("vol://...")

df.write.saveAsTable()

# Databricks DLT（01_source/dlt/01_bronze.py） @dlt.table(name="bronze_sales") def bronze_sales(): return spark.readStream.format("cloudFiles").option("cloudFiles.format","csv").load(VOL_PATH)

# ZettaPark（03_lakehouse/01_bronze.py）— 最小化改写 from clickzetta.zettapark import Session # ← import pyspark → clickzetta.zettapark session = Session.builder.configs({...}).create() # ← spark 全局注入 → 显式创建 df = session.read.option("header","true").csv("vol://apparel_bronze.raw_data/sales.csv") # readStream.format("cloudFiles")... → session.read.csv("vol://...") df.write.mode("overwrite").saveAsTable("apparel_bronze.raw_sales") # @dlt.table(name=X) → df.write.saveAsTable(X)（函数体最后一行）

第二步：

@dlt.expect_or_drop

@dlt.expect_or_drop

→

df.filter()

df.filter()

DLT 数据质量约束直接转为 ZettaPark

.filter()

# Databricks DLT（02A_silver.py） @dlt.view(name="customers_cleaned_stream") @dlt.expect_or_drop("valid_customer_id", "customer_id IS NOT NULL") @dlt.expect("realistic_age", "age >= 18 AND age <= 100") def customers_cleaned_stream(): return dlt.read_stream(f"LIVE.{BRONZE_CUSTOMERS}")

# ZettaPark（03_lakehouse/02B_silver.py） customers = session.table("apparel_bronze.raw_customers") # dlt.read_stream → session.table() customers = (customers .filter(F.col("customer_id").isNotNull()) # @dlt.expect_or_drop("valid_customer_id",...) .filter((F.col("age") >= 18) & (F.col("age") <= 100)) # @dlt.expect("realistic_age",...) )

第三步：SCD Type 2 —

dlt.create_auto_cdc_flow

dlt.create_auto_cdc_flow

→

LEAD()

LEAD()

窗口函数

create_auto_cdc_flow(stored_as_scd_type=2)

F.lead()

Window

# Databricks DLT（02B_silver.py） dlt.create_auto_cdc_flow( target=SILVER_CUSTOMERS, source=f"live.{CUSTOMERS_CLEANED_STREAM}", keys=["customer_id"], sequence_by=F.col("last_update_time"), stored_as_scd_type=2, )

# ZettaPark（03_lakehouse/02B_silver.py） from clickzetta.zettapark.window import Window # ← pyspark.sql.window → clickzetta.zettapark.window w = Window.partitionBy("customer_id").orderBy("last_update_time") # F.lead() 和 Window API 完全一致，无需改动 silver_customers = customers.withColumn( "__end_at", F.lead("last_update_time").over(w) # SCD2 end timestamp ).withColumn( "__is_current", F.lead("last_update_time").over(w).isNull() ).withColumn( "__end_at", F.coalesce(F.col("__end_at"), F.lit("9999-12-31 23:59:59").cast(TimestampType())) ) silver_customers.write.mode("overwrite").saveAsTable("apparel_silver.silver_customers")

结果：150 条原始记录 → 150 行（含 30 条历史）→ 120 条 current 快照（

__is_current = TRUE

第四步：时间窗口聚合 —

F.window()

F.window()

→

F.to_date()

F.to_date()

F.window("event_time","1 day")

F.to_date()

# Databricks DLT（03_gold.py） df.groupBy(F.window("event_time","1 day").alias("sale_window"), "store_id","store_name") \ .agg(F.round(F.sum("total_amount"),2).alias("total_revenue")) \ .select(F.col("sale_window.start").cast("date").alias("sale_date"), ...)

# ZettaPark（03_lakehouse/03_gold.py） facts.withColumn("sale_date", F.to_date(F.col("event_time"))) # F.window → F.to_date .groupBy("sale_date","store_id","store_name") .agg(F.round(F.sum("total_amount"),2).alias("total_revenue"), F.count("transaction_id").alias("total_transactions"), ...)

F.to_date()

F.window("1 day")

纯 SQL 替代方案

如果团队偏向 SQL，

03_lakehouse/sql/

cz-cli sql --file 03_lakehouse/sql/02_silver.sql --profile aws_singapore_prod --sync --write cz-cli sql --file 03_lakehouse/sql/03_gold.sql --profile aws_singapore_prod --sync --write

@dlt.expect_or_drop

WHERE condition

create_auto_cdc_flow(scd=2)

LEAD() OVER (PARTITION BY key ORDER BY seq)

F.window("event_time","1 day")

DATE_TRUNC('day', event_time)

ZettaPark 是主推路径（代码改动最小，保留 PySpark 技能）；SQL 适合 SQL-first 团队或快速验证。

方案 C：Dynamic Table（GIC）— DLT 的原生对标

Dynamic Table 是 DLT

@dlt.table

REFRESH INTERVAL

# Databricks DLT — 声明式管道 @dlt.table(name="gold_daily_sales_by_store") def gold_daily_sales_by_store(): return df.groupBy(F.window("event_time","1 day"), "store_id","store_name") \ .agg(F.sum("total_amount").alias("total_revenue"))

-- Lakehouse Dynamic Table — 直接对标（03_lakehouse/dynamic_tables/gold_dynamic_tables.sql） CREATE OR REPLACE DYNAMIC TABLE apparel_gold.dt_daily_sales_by_store REFRESH INTERVAL 10 MINUTE -- DLT 按新数据触发；DT 按时间调度 VCLUSTER DEFAULT AS SELECT CAST(event_time AS DATE) AS sale_date, store_id, store_name, ROUND(SUM(total_amount), 2) AS total_revenue, ... FROM apparel_gold.denormalized_sales_facts GROUP BY CAST(event_time AS DATE), store_id, store_name; -- 手动触发（等价于 DLT 的 pipeline trigger） REFRESH DYNAMIC TABLE apparel_gold.dt_daily_sales_by_store;

@dlt.table(name=X)

CREATE DYNAMIC TABLE X ... AS SELECT ...

REFRESH INTERVAL N MINUTE

F.window("1 day")

CAST(event_time AS DATE)

create_auto_cdc_flow(scd=2)

LEAD() OVER Window

实测在 AWS 新加坡创建并刷新了 4 个 Dynamic Table，全部通过 e2e 验证：

dt_customers_current

dt_daily_sales_by_store

dt_product_performance

dt_customer_lifetime_value

关于 Pipe（Auto Loader 对标）

Pipe 是 Databricks Auto Loader 的对标——持续监听对象存储（OSS/S3/COS）的新文件，自动触发 COPY INTO 导入。已在 AWS 新加坡实测通过：上传新 CSV 到 S3，Pipe 在 15 秒内自动摄取。

-- Step 1: 创建外部 Volume（连接到 S3/OSS/COS） CREATE EXTERNAL VOLUME apparel_bronze.s3_sales_landing LOCATION 's3://your-bucket/apparel_landing/' USING CONNECTION s3_conn DIRECTORY = (enable=true, auto_refresh=true) RECURSIVE = true; -- Step 2: 建表（外部 Volume 不支持 inferSchema，需显式定义） CREATE TABLE apparel_bronze.raw_sales ( transaction_id BIGINT, store_id BIGINT, event_time STRING, customer_id BIGINT, product_id BIGINT, quantity BIGINT, unit_price DOUBLE, total_amount DOUBLE, payment_method STRING, discount_applied DOUBLE, tax_amount DOUBLE ); -- Step 3: 创建 Pipe -- DLT: spark.readStream.format("cloudFiles").option("cloudFiles.format","csv").load(PATH) CREATE OR REPLACE PIPE apparel_bronze.pipe_sales VIRTUAL_CLUSTER = 'DEFAULT' INGEST_MODE = 'LIST_PURGE' -- 轮询扫描新文件，等同 Auto Loader LIST 模式 AS COPY INTO apparel_bronze.raw_sales FROM VOLUME apparel_bronze.s3_sales_landing USING CSV OPTIONS ('header'='true') PURGE = TRUE -- 处理后删除 landing zone 文件 ON_ERROR = CONTINUE;

e2e 验证结果

在 AWS 新加坡实例（

aws_singapore_prod

注意事项

• 流式 vs 批量：Auto Loader 持续监听新文件，COPY INTO 是一次性批量。生产环境用 Lakehouse Pipe 替代 Auto Loader——配置后自动摄取新文件，无需手动触发。

• @dlt.expect

  @dlt.expect
  的 warn 语义：

  expect

  expect
  （warn 级）在 DLT 里会计数但不丢弃记录。迁移到 SQL 时，可以选择不过滤（完全忽略），或者加

  WHERE

  WHERE
  过滤（等价于升级为

  expect_or_drop

  expect_or_drop
  ）。根据业务需求决定。
• DT 自动增量：Lakehouse Dynamic Table 定义后会自动增量刷新（类似 DLT 的流式更新），不需要显式写

  read_stream

  read_stream
  。
• SCD2

  __is_current

  __is_current
  字段：

  LEAD()

  LEAD()
  返回 NULL 表示没有后续版本，即为当前记录。

  COALESCE(__end_at, '9999-12-31')

  COALESCE(__end_at, '9999-12-31')
  是 SCD2 标准惯用法，两边行为完全一致。

相关文档

Dynamic Table (GIC)

• 动态表概述：声明式增量计算原理与架构
• 动态表 SQL 手册：CREATE DYNAMIC TABLE 完整语法
• 动态表使用指南：批流一体管道示例

其他迁移实战

• Databricks Notebook → Lakehouse 迁移实战（零售 Medallion 管道）
• dbt-databricks → dbt-clickzetta 迁移实战（金融支付管道）
• Spark 迁移指南：Spark 生态迁移概览