Snowflake中源表到目标表的增量同步最优方案咨询

针对你遇到的源表定期截断重加载、流+Merge无法捕捉删除操作的问题，以下是几个落地性强的解决方案，按适用场景排序：

方案1：新增快照+变更日志层（推荐用于大数据量、高频同步场景） #

核心思路是在源表截断前留存历史数据快照，通过对比新旧快照生成完整的增删改信号，再用变更日志同步目标表，完全规避截断导致的删除信号丢失问题。

操作步骤： #

1. 预加载快照：在每次截断源表前，将当前源表数据备份到快照表（带时间戳标记）

-- 创建快照表（仅需执行一次）
CREATE TABLE IF NOT EXISTS source_table_snapshot (
  id INT PRIMARY KEY,
  col1 VARCHAR(50),
  col2 INT,
  load_timestamp TIMESTAMP NOT NULL
);

-- 加载新数据前，备份当前源表到快照表
INSERT INTO source_table_snapshot (id, col1, col2, load_timestamp)
SELECT id, col1, col2, CURRENT_TIMESTAMP()
FROM source_table;

2. 执行原加载流程：截断源表并加载新数据

TRUNCATE TABLE source_table;
-- 这里替换成你的实际加载逻辑，比如LOAD DATA/INSERT FROM SELECT等
LOAD DATA INFILE '/path/to/new_data.csv' INTO TABLE source_table;

3. 生成变更日志：对比新旧快照，输出增删改记录

-- 创建变更日志表（仅需执行一次）
CREATE TABLE IF NOT EXISTS change_log (
  id INT PRIMARY KEY,
  col1 VARCHAR(50),
  col2 INT,
  operation_type ENUM('INSERT', 'UPDATE', 'DELETE') NOT NULL,
  change_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP()
);

-- 清空上一次的变更日志（按需）
TRUNCATE TABLE change_log;

-- 插入新增记录
INSERT INTO change_log (id, col1, col2, operation_type)
SELECT s.id, s.col1, s.col2, 'INSERT'
FROM source_table s
LEFT JOIN source_table_snapshot snap ON s.id = snap.id
WHERE snap.id IS NULL;

-- 插入更新记录
INSERT INTO change_log (id, col1, col2, operation_type)
SELECT s.id, s.col1, s.col2, 'UPDATE'
FROM source_table s
JOIN source_table_snapshot snap ON s.id = snap.id
WHERE s.col1 != snap.col1 OR s.col2 != snap.col2;

-- 插入删除记录
INSERT INTO change_log (id, col1, col2, operation_type)
SELECT snap.id, snap.col1, snap.col2, 'DELETE'
FROM source_table_snapshot snap
LEFT JOIN source_table s ON snap.id = s.id
WHERE s.id IS NULL;

4. 同步目标表：用变更日志执行Merge操作

MERGE INTO target_table t
USING change_log cl ON t.id = cl.id
WHEN MATCHED AND cl.operation_type = 'UPDATE' THEN
  UPDATE SET t.col1 = cl.col1, t.col2 = cl.col2
WHEN MATCHED AND cl.operation_type = 'DELETE' THEN
  DELETE
WHEN NOT MATCHED AND cl.operation_type = 'INSERT' THEN
  INSERT (id, col1, col2) VALUES (cl.id, cl.col1, cl.col2);

优势： #

• 精准捕捉所有变更（包括删除），同步逻辑稳定
• 增量同步，性能优于全量对比

不足： #

• 需要额外存储快照表和变更日志表，增加少量存储成本

方案2：直接全量对比源表与目标表（适合小数据量、低频同步场景） #

如果源表数据量不大，无需额外中间层，直接对比源表和目标表的差异，执行对应的增删改操作。

操作代码： #

-- 先执行更新操作
UPDATE target_table t
JOIN source_table s ON t.id = s.id
SET t.col1 = s.col1, t.col2 = s.col2
WHERE t.col1 != s.col1 OR t.col2 != s.col2;

-- 执行新增操作
INSERT INTO target_table (id, col1, col2)
SELECT id, col1, col2
FROM source_table s
LEFT JOIN target_table t ON s.id = t.id
WHERE t.id IS NULL;

-- 执行删除操作
DELETE t
FROM target_table t
LEFT JOIN source_table s ON t.id = s.id
WHERE s.id IS NULL;

优势： #

• 实现简单，无需额外表结构
• 无需修改现有源表加载流程

不足： #

• 全量对比在数据量大时性能差，同步耗时久

方案3：修改源表加载逻辑，避免截断（若上游流程可调整） #

如果能调整源表的加载方式，将「截断+全量加载」改为「临时表替换」，可以从源头避免截断导致的变更信号丢失问题。

操作步骤： #

1. 将新数据加载到临时表source_temp

CREATE TEMPORARY TABLE source_temp LIKE source_table;
LOAD DATA INFILE '/path/to/new_data.csv' INTO TABLE source_temp;

2. 对比临时表和源表，生成变更日志（同方案1的步骤3），同步目标表

3. 原子性替换源表

-- 原子替换，避免源表空窗口
RENAME TABLE source_table TO source_old, source_temp TO source_table;
DROP TABLE source_old;

优势： #

• 源表更新原子性，无空窗口
• 可直接用流处理或Merge同步，逻辑更简洁

不足： #

• 需要修改上游数据加载流程，可能涉及多系统协调

方案选择建议 #

• 大数据量、高同步频率：选方案1
• 小数据量、低同步频率：选方案2
• 上游流程可调整：优先选方案3

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