Snowflake至Delta Lake增量同步的更优实现方案咨询

你的核心问题在于当前管道仅复制了源表的行数据，缺少变更类型（插入/更新/删除）、变更顺序标识等关键元数据，导致无法还原源表的实际状态。以下是两种更优的解决方案：

方案一：增强现有管道，补充CDC元数据 #

无需更换整体架构，只需在Snowflake侧补充变更元数据，再在Databricks侧利用这些元数据构建可还原的Delta表。

步骤1：修改Snowflake Task，包含变更元数据 #

Snowflake Stream内置系统元数据列，可直接在COPY INTO时引入：

CREATE OR REPLACE TASK UNLOAD_TASK 
WAREHOUSE = COMPUTE_WH
AS 
COPY INTO @my_external_stage
FROM (
  SELECT 
    *,
    METADATA$ACTION AS change_action, -- 标识变更类型：INSERT/DELETE
    METADATA$ISUPDATE AS is_update,   -- 是否为更新操作（仅UPDATE时为TRUE）
    METADATA$ROW_ID AS row_id,        -- 源表行的唯一标识，用于跟踪变更
    CURRENT_TIMESTAMP() AS ingestion_timestamp -- 任务执行时间，用于排序变更
  FROM BOOKS_STREAM
) 
-- 改用Parquet格式替代CSV，更好保留数据类型与元数据
FILE_FORMAT = (TYPE='PARQUET' COMPRESSION = SNAPPY)
INCLUDE_QUERY_ID = TRUE;

步骤2：在Databricks DLT中处理CDC数据 #

先构建Bronze层存储所有原始变更，再用apply_changes构建Silver层还原源表状态：

Bronze层（存储全量变更）

@dlt.table("books_bronze", temporary=False)
def books_bronze_ingest():
  return (spark
    .readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    .option("cloudFiles.schemaLocation", schemaPath)
    .load("<same-path-as-external-stage>")
  )

Silver层（还原源表最新状态）

利用DLT的apply_changes自动处理插入、更新、删除：

@dlt.table("books_silver", temporary=False)
def books_silver():
    return dlt.apply_changes(
        target="books_silver",
        source="books_bronze",
        keys=["id"],  -- 源表主键，用于匹配行
        sequence_by="ingestion_timestamp", -- 变更顺序依据，确保操作顺序正确
        apply_as_deletes="change_action = 'DELETE'", -- 标记删除操作
        apply_as_truncates=None,
        column_list="*",
        except_column_list=["change_action", "is_update", "row_id", "ingestion_timestamp"] -- 排除元数据列
    )

方案二：使用Snowpipe Streaming直接推送结构化变更 #

如果你的Snowflake版本支持，推荐使用Snowpipe Streaming替代Task+COPY INTO，它能更实时地将变更数据以结构化格式（如Parquet/JSON）写入云存储，且内置CDC元数据，无需手动拼接：

步骤1：创建Snowpipe Streaming管道 #

CREATE OR REPLACE PIPE BOOKS_STREAM_PIPE
AUTO_INGEST = TRUE
AS
COPY INTO @my_external_stage
FROM (
  SELECT 
    *,
    METADATA$ACTION,
    METADATA$ISUPDATE,
    METADATA$ROW_ID
  FROM BOOKS_STREAM
)
FILE_FORMAT = (TYPE='PARQUET' COMPRESSION = SNAPPY);

步骤2：Databricks侧处理流程 #

与方案一的DLT代码一致，利用apply_changes构建Silver层即可。Snowpipe Streaming会自动按变更顺序写入文件，确保元数据完整性。

关键优势说明 #

1. 元数据完整性：通过引入Snowflake Stream的系统列，完整保留变更类型、顺序等信息，可精准还原源表状态。
2. 数据类型一致性：改用Parquet格式替代CSV，避免类型丢失、空值处理等问题，提升数据质量。
3. 自动化CDC处理：DLT的apply_changes简化了变更合并逻辑，无需手动编写复杂的merge语句。

内容的提问来源于stack exchange，提问作者TheRealJimShady