使用Azure Data Factory实现PostgreSQL增量数据复制的方案咨询

方案一：临时表+合并语句（最通用） #

• 核心思路：先把增量数据落地到目标库的临时表，再通过PostgreSQL原生语句完成Upsert逻辑
  1. 复制活动配置：将增量数据写入目标库的临时表（比如tmp_inc_data），Sink选择追加写入模式
  2. 添加脚本活动执行合并操作：
     • 若目标库是PostgreSQL 15及以上，直接用MERGE：

       MERGE INTO target_table t
       USING tmp_inc_data s
       ON t.id = s.id -- 替换成你的主键字段
       WHEN MATCHED THEN
         UPDATE SET name = s.name, update_time = s.update_time -- 列出需要更新的字段
       WHEN NOT MATCHED THEN
         INSERT (id, name, update_time) VALUES (s.id, s.name, s.update_time);
       

     • 若版本低于15，用INSERT ... ON CONFLICT替代：

       INSERT INTO target_table (id, name, update_time)
       SELECT id, name, update_time FROM tmp_inc_data
       ON CONFLICT (id) DO UPDATE
       SET name = EXCLUDED.name, update_time = EXCLUDED.update_time;
       

  3. 可选：添加脚本活动清理临时表：DROP TABLE IF EXISTS tmp_inc_data;

方案二：删除旧数据+追加新数据 #

• 适用场景：增量数据集包含需要更新的旧记录（即水印列更新过的记录）
  1. 用Lookup活动获取本次增量数据的主键列表，或者根据水印列筛选目标库中待更新的主键
  2. 脚本活动删除目标库中对应旧记录：

     DELETE FROM target_table WHERE id IN ('1001', '1002', '1003'); -- 替换成实际主键集合
     

     注意：主键数量多的话，建议先把主键写入临时表，再关联删除，避免SQL语句过长
  3. 执行复制活动，将增量数据追加写入目标库

方案三：触发器+CDC变更捕获 #

• 适用场景：需要准实时同步，或不想依赖水印列判断增量
  1. 在源库创建触发器，把所有INSERT/UPDATE/DELETE操作记录到专门的CDC日志表（比如source_cdc_log），日志表要包含变更类型、变更时间、完整记录数据
  2. 定期从CDC日志表同步增量数据到目标库，根据变更类型执行对应操作：
     • 新增/更新：用INSERT ... ON CONFLICT DO UPDATE
     • 删除：直接删除目标库对应主键的记录
  3. 同步完成后标记日志表中已处理的记录（比如加个is_processed字段），避免重复同步

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