Snowflake Dynamic Tables 迁移实战：Bronze–Silver–Gold 三层管道

如果你在 Snowflake 上用 Dynamic Tables 构建了 Medallion 数据管道，迁移到 ClickZetta Lakehouse 的成本主要集中在 4 处语法差异，核心的 SQL 查询逻辑不需要改动。

本文用一个真实迁移项目演示完整过程：将基于 Snowflake Dynamic Tables 的 Bronze–Silver–Gold 三层管道迁移到 ClickZetta Lakehouse，数据源使用两个平台共有的 TPC-H 标准数据集，全部 SQL 经过 cz-cli 实际验证。

完整代码见 GitHub：snowflake2lakehouse-dynamic-tables

原始项目

snowflake2lakehouse-dynamic-tables 基于 Techy-Malay/snowflake-bsg-dynamic-tables 改造，演示如何在 Snowflake 上用 Dynamic Tables 实现 Bronze–Silver–Gold 三层架构。项目以 TPC-H ORDERS 表为数据源，经过三层 Dynamic Table 处理后输出每日销售汇总。

迁移后的代码在

03_lakehouse/

01_snowflake/

结论先行

核心 SQL 查询逻辑不需要改动。 4 处改动全是平台配置替换：

TARGET_LAG

REFRESH INTERVAL

WAREHOUSE

VCLUSTER

DATA_RETENTION_TIME_IN_DAYS

DOWNSTREAM

TARGET_LAG

REFRESH INTERVAL

WAREHOUSE

VCLUSTER

REFRESH INTERVAL

ALTER TABLE SET PROPERTIES

SNOWFLAKE_SAMPLE_DATA.TPCH_SF1

clickzetta_sample_data.tpch_100g

清洗、去重（QUALIFY）、聚合、日期截断——核心 SQL 逻辑语法完全一致，不需要改。

技术栈对比

WAREHOUSE = compute_wh

VCLUSTER default

TARGET_LAG = '5 minutes'

REFRESH INTERVAL '5' MINUTE

TARGET_LAG = 'DOWNSTREAM'

ALTER DYNAMIC TABLE ... REFRESH

REFRESH DYNAMIC TABLE ...

DATA_RETENTION_TIME_IN_DAYS = 1

ALTER TABLE ... SET PROPERTIES ('data_retention_days' = '1')

SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.ORDERS

clickzetta_sample_data.tpch_100g.orders

USE SCHEMA

QUALIFY ROW_NUMBER() OVER (...) = 1

DATE_TRUNC('day', ts)

变化的主要是平台配置——从 Snowflake Virtual Warehouse 换成 Lakehouse VCluster，从

TARGET_LAG

REFRESH INTERVAL

架构概览

SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.ORDERS clickzetta_sample_data.tpch_100g.orders （Snowflake 内置样本数据集，1GB） （Lakehouse 共享数据集，100GB） │ │ ▼ ▼ ORDERS_STG orders_stg DATA_RETENTION_TIME_IN_DAYS = 1 SET PROPERTIES ('data_retention_days' = '1') │ │ ▼ TARGET_LAG = '5 minutes' ▼ REFRESH INTERVAL '5' MINUTE bronze_orders bronze_orders （原始数据 + 摄取元数据） （原始数据 + 摄取元数据） │ │ ▼ TARGET_LAG = '5 minutes' + QUALIFY ▼ REFRESH INTERVAL '5' MINUTE + QUALIFY silver_orders silver_orders （清洗、去重、类型标准化） （清洗、去重、类型标准化） │ │ ▼ TARGET_LAG = '10 minutes' ▼ REFRESH INTERVAL '10' MINUTE gold_sales_summary gold_sales_summary （每日销售汇总，面向分析） （每日销售汇总，面向分析）

预览

迁移步骤

第一步：替换样本数据集引用

Snowflake 的内置样本数据通过

SNOWFLAKE_SAMPLE_DATA

clickzetta_sample_data

Snowflake：

CREATE OR REPLACE TABLE ARCH_BSG_DYNAMIC_TABLES.ORDERS_STG DATA_RETENTION_TIME_IN_DAYS = 1 AS SELECT O_ORDERKEY AS order_id, O_CUSTKEY AS customer_id, O_ORDERDATE AS order_ts, O_TOTALPRICE AS amount FROM SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.ORDERS;

Lakehouse：

CREATE SCHEMA IF NOT EXISTS bsg_dynamic_tables; CREATE OR REPLACE TABLE bsg_dynamic_tables.orders_stg AS SELECT O_ORDERKEY AS order_id, O_CUSTKEY AS customer_id, O_ORDERDATE AS order_ts, O_TOTALPRICE AS amount FROM clickzetta_sample_data.tpch_100g.orders; ALTER TABLE bsg_dynamic_tables.orders_stg SET PROPERTIES ('data_retention_days' = '1');

两处变化：

1. SNOWFLAKE_SAMPLE_DATA.TPCH_SF1

   SNOWFLAKE_SAMPLE_DATA.TPCH_SF1
   →

   clickzetta_sample_data.tpch_100g

   clickzetta_sample_data.tpch_100g
   （数据集名称不同，列名相同）

2. DATA_RETENTION_TIME_IN_DAYS = 1

   DATA_RETENTION_TIME_IN_DAYS = 1
   从 DDL 内联选项改为建表后单独执行

   ALTER TABLE ... SET PROPERTIES

   ALTER TABLE ... SET PROPERTIES

第二步：替换 Dynamic Table 语法

这是迁移的核心部分，涉及 3 处语法变化。

TARGET_LAG

REFRESH INTERVAL

Snowflake 用

TARGET_LAG

REFRESH INTERVAL

Snowflake：

CREATE OR REPLACE DYNAMIC TABLE bronze_orders TARGET_LAG = '5 minutes' WAREHOUSE = compute_wh AS SELECT ...

Lakehouse：

CREATE OR REPLACE DYNAMIC TABLE bsg_dynamic_tables.bronze_orders REFRESH INTERVAL '5' MINUTE VCLUSTER default AS SELECT ...

WAREHOUSE

VCLUSTER

Snowflake Virtual Warehouse 对应 Lakehouse VCluster，填写你的 VCluster 名称（大多数实例默认为

default

TARGET_LAG = 'DOWNSTREAM'

Snowflake 支持

TARGET_LAG = 'DOWNSTREAM'

REFRESH INTERVAL

实践建议：将 Tier 1（Bronze/Silver）的刷新间隔设置得比 Tier 2（Gold）短，近似模拟级联效果。例如 Bronze/Silver 设 5 分钟，Gold 设 10 分钟，Gold 刷新时 Bronze/Silver 已经是最新数据。

第三步：替换手动刷新命令

Snowflake：

ALTER DYNAMIC TABLE bronze_orders REFRESH;

Lakehouse：

REFRESH DYNAMIC TABLE bsg_dynamic_tables.bronze_orders;

注意刷新必须按依赖顺序手动执行——Lakehouse 不会自动级联：

REFRESH DYNAMIC TABLE bsg_dynamic_tables.bronze_orders; REFRESH DYNAMIC TABLE bsg_dynamic_tables.silver_orders; REFRESH DYNAMIC TABLE bsg_dynamic_tables.gold_sales_summary;

第四步：完全兼容的部分（无需修改）

以下语法在 Lakehouse 中与 Snowflake 完全一致：

QUALIFY ROW_NUMBER() OVER (...) = 1

-- Snowflake 和 Lakehouse 写法完全相同 FROM bronze_orders QUALIFY ROW_NUMBER() OVER ( PARTITION BY order_id ORDER BY ingestion_ts DESC ) = 1;

DATE_TRUNC('day', ts)

-- Snowflake 和 Lakehouse 写法完全相同 SELECT DATE_TRUNC('day', order_ts) AS order_date, COUNT(*) AS total_orders, SUM(amount) AS total_sales FROM silver_orders GROUP BY DATE_TRUNC('day', order_ts);

CAST(order_ts AS TIMESTAMP)

-- Snowflake 和 Lakehouse 写法完全相同 CAST(order_ts AS TIMESTAMP) AS order_ts

验证结果

所有 SQL 经过 cz-cli 在 Lakehouse 实例上实际运行验证：

orders_stg

bronze_orders

ingestion_ts

source_system

silver_orders

gold_sales_summary

运行完成后，用以下命令清理所有 Lakehouse 对象：

cz-cli sql -f 03_lakehouse/06_cleanup.sql --profile <your-profile> --sync --write

迁移结论

Snowflake Dynamic Tables 与 Lakehouse Dynamic Tables 的 SQL 查询逻辑高度兼容，本项目验证了以下结论：

完全兼容（无需修改）：

• QUALIFY ROW_NUMBER() OVER (...) = 1

  QUALIFY ROW_NUMBER() OVER (...) = 1
  去重

• DATE_TRUNC('day', ts)

  DATE_TRUNC('day', ts)
  日期截断

• CAST(col AS TYPE)

  CAST(col AS TYPE)
  类型转换
• 标准聚合函数：

  COUNT

  COUNT
  、

  SUM

  SUM
  、

  AVG

  AVG

• CURRENT_TIMESTAMP()

  CURRENT_TIMESTAMP()
  系统函数

需要修改的 4 处：

WAREHOUSE = wh_name

VCLUSTER vcluster_name

TARGET_LAG = 'N minutes'

REFRESH INTERVAL 'N' MINUTE

TARGET_LAG = 'DOWNSTREAM'

ALTER DYNAMIC TABLE ... REFRESH

REFRESH DYNAMIC TABLE ...

DATA_RETENTION_TIME_IN_DAYS = N

ALTER TABLE ... SET PROPERTIES ('data_retention_days' = 'N')

参考

• GitHub 项目：snowflake2lakehouse-dynamic-tables
• 原始项目：Techy-Malay/snowflake-bsg-dynamic-tables
• 动态表简介
• CREATE DYNAMIC TABLE
• Time Travel
• 动态表开发入门