如何使用Snowflake动态表实现预加载删除及指定数据加载场景

Snowflake动态表是声明式同步工具，无法直接支持"先删除目标表特定记录再加载新数据"的有状态操作，以下是适配你场景的组合解决方案：

1. 用动态表创建Stage Table 1和Stage Table 2 #

动态表适合自动同步源表关联后的结果，保持stage表实时更新：

-- 创建Stage Table 1（关联源表集合set1）
CREATE OR REPLACE DYNAMIC TABLE stage_table_1
TARGET_LAG = '1 MINUTE' -- 按需设置延迟，最小1分钟
WAREHOUSE = your_warehouse_name
AS
SELECT 
  s1.id, s1.col1, s2.col2 -- 按需选择字段
FROM set1_table_1 s1
JOIN set1_table_2 s2 ON s1.id = s2.id
-- 补充set1集合内其他表的关联逻辑
;

-- 创建Stage Table 2（关联源表集合set2）
CREATE OR REPLACE DYNAMIC TABLE stage_table_2
TARGET_LAG = '1 MINUTE'
WAREHOUSE = your_warehouse_name
AS
SELECT 
  s2a.key, s2a.col_a, s2b.col_b, 'external' AS record_type -- 明确记录类型标识
FROM set2_table_a s2a
JOIN set2_table_b s2b ON s2a.key = s2b.key
-- 补充set2集合内其他表的关联逻辑
;

2. 创建任务处理删除+加载逻辑 #

用Snowflake任务执行"删除目标表external记录→加载Stage Table 2数据"的 imperative 操作，同时可依赖动态表刷新确保数据一致性：

-- 创建核心任务
CREATE OR REPLACE TARGET task_load_to_target
WAREHOUSE = your_warehouse_name
-- 可选：依赖Stage Table 2刷新完成后自动触发，替代定时调度
AFTER stage_table_2
-- 若需定时调度，替换为：SCHEDULE = 'USING CRON 0 * * * * UTC'
AS
BEGIN
  -- 步骤3：删除目标表中record_type为external的记录
  DELETE FROM target_table WHERE record_type = 'external';
  
  -- 步骤4：加载Stage Table 2数据到目标表
  INSERT INTO target_table (id, col1, col2, record_type)
  SELECT key, col_a, col_b, record_type FROM stage_table_2;
END;

-- 启用任务
ALTER TASK task_load_to_target RESUME;

替代方案：流+任务全流程处理 #

若无需保留stage表的持久化数据，可直接用流捕获源表变化，任务一站式处理关联、删除、加载：

-- 为set2核心表创建流（捕获增量变化）
CREATE OR REPLACE STREAM set2_stream ON TABLE set2_table_a;

-- 创建全流程任务
CREATE OR REPLACE TASK task_full_load
WAREHOUSE = your_warehouse_name
WHEN SYSTEM$STREAM_HAS_DATA('set2_stream') -- 仅当源表有变化时触发
AS
BEGIN
  -- 临时生成stage1数据（按需使用）
  CREATE OR REPLACE TEMPORARY TABLE temp_stage1 AS
  SELECT s1.id, s1.col1, s2.col2 FROM set1_table_1 s1 JOIN set1_table_2 s2 ON s1.id = s2.id;
  
  -- 临时生成stage2数据
  CREATE OR REPLACE TEMPORARY TABLE temp_stage2 AS
  SELECT s2a.key, s2a.col_a, s2b.col_b, 'external' AS record_type 
  FROM set2_stream s2a JOIN set2_table_b s2b ON s2a.key = s2b.key;
  
  -- 删除目标表external记录
  DELETE FROM target_table WHERE record_type = 'external';
  
  -- 加载stage2数据
  INSERT INTO target_table SELECT * FROM temp_stage2;
END;

ALTER TASK task_full_load RESUME;

关键注意事项 #

• 动态表仅负责无状态的映射/同步，有状态的删除操作必须通过任务实现
• 任务依赖动态表的AFTER配置，可确保stage表数据完全刷新后再执行加载，避免数据不一致
• 若需增量处理，配合流（Stream）可大幅减少全量扫描的资源开销

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