Azure Data Pipeline实现Snowflake无Stream的CDC数据复制故障求助

核心原因 #

Azure Copy Activity的Snowflake数据源仅支持单条SELECT查询语句，不允许混合执行SET变量、UPDATE等非查询操作，也无法识别Snowflake会话级变量（如$ts1）——因为每个Copy Activity的数据库连接是独立会话，变量无法跨会话生效。你原来的脚本包含多语句操作，这是导致失败的直接原因。

分步解决方案 #

1. 拆分操作流程 #

将原脚本的三个逻辑拆分为独立活动：

• 读取上次同步时间（Lookup活动）
• 执行CDC数据查询与复制（Copy Activity）
• 更新同步时间（Snowflake活动或存储过程活动）

2. 读取同步时间（Lookup活动） #

在For Each活动内，先添加Lookup活动（命名为GetLastRuntime），执行单条查询获取当前表的上次同步时间：

SELECT LASTRUNTIME FROM CDC_TABLE WHERE TABLE_NAME = '@{item().TABLE_NAME}'

这里@{item().TABLE_NAME}是For Each遍历的表名动态变量。

3. 调整Copy Activity查询语句 #

将CDC查询改为纯SELECT语句，用动态内容替换原来的会话变量和硬编码表名：

SELECT *
EXCLUDE (METADATA$ACTION, METADATA$ISUPDATE, METADATA$ROW_ID)
,CASE WHEN METADATA$ACTION = 'INSERT' AND METADATA$ISUPDATE = 'TRUE' THEN 'U'
      WHEN METADATA$ACTION = 'INSERT' AND METADATA$ISUPDATE = 'FALSE' THEN 'I'
      ELSE 'D'
 END AS CHANGE_OPERATION
FROM (
    SELECT * 
    FROM @{item().TABLE_NAME} CHANGES(INFORMATION => DEFAULT) 
    AT (timestamp => '@{activity('GetLastRuntime').output.firstRow.LASTRUNTIME}')
)
WHERE METADATA$ACTION != 'DELETE'
ORDER BY AGENT_CODE

• 用@{item().TABLE_NAME}动态替换表名
• 用@{activity('GetLastRuntime').output.firstRow.LASTRUNTIME}传递从Lookup获取的同步时间
• 确保整个查询是单条SELECT，无其他非查询语句

4. 更新同步时间（独立活动） #

在Copy Activity执行完成后，添加Snowflake活动（或存储过程活动）执行UPDATE语句：

UPDATE CDC_TABLE 
SET PREVIOUS_RUNTIME = LASTRUNTIME, LASTRUNTIME = CURRENT_TIMESTAMP
WHERE TABLE_NAME = '@{item().TABLE_NAME}'

额外注意事项 #

• 确保目标表已开启数据历史：ALTER TABLE @{item().TABLE_NAME} SET DATA_RETENTION_TIME_IN_DAYS = 7;（可根据业务需求调整天数）
• 同步时间的格式需符合Snowflake要求：Lookup返回的时间戳需是Snowflake可识别的格式（如YYYY-MM-DD HH:MI:SS.FF），若格式不符可在Lookup查询中用TO_TIMESTAMP(LASTRUNTIME)转换。

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