使用Copy Activity同步PostgreSQL到Azure Gen2的增量加载重复ID与Upsert难题

针对你用Copy Activity基于时间戳增量抽取PostgreSQL数据到ADLS Gen2（Parquet格式）时出现重复ID，且无法直接Upsert的问题，给你几个可行的方案：

方案1：用Data Flow实现Upsert逻辑 #

Data Flow支持对Parquet数据集的Upsert操作，步骤如下：

• 配置两个数据源：一个是PostgreSQL的增量数据（用时间戳过滤WHERE update_time > @last_run_time），另一个是ADLS Gen2中已有的Parquet数据。
• 添加Join组件，以ID作为关联键，将增量数据和现有数据进行左连接。
• 添加Derived Column组件，处理字段更新逻辑：比如对每个字段，用iif(isNull(增量数据.字段), 现有数据.字段, 增量数据.字段)保留最新值，或直接用增量数据覆盖。
• 配置Sink为ADLS Gen2的Parquet数据集，在Sink设置中选择Upsert，指定ID作为Upsert键，同时可开启Optimize for Upserts提升性能。

方案2：通过Spark作业做去重合并 #

借助Synapse Spark或Azure Databricks处理重复数据，步骤：

1. 先用Copy Activity将增量数据写入ADLS Gen2的临时路径（比如temp/incremental/）。
2. 提交Spark作业：
   • 读取临时路径的增量Parquet数据，以及目标路径的现有Parquet数据，合并成一个DataFrame。
   • 按ID分组，保留每组中update_time最大的记录（即最新版本的数据）：

     from pyspark.sql import functions as F
     
     merged_df = combined_df.groupBy("ID") \
         .agg(F.max("update_time").alias("latest_time")) \
         .join(combined_df, on=["ID", "update_time"], how="inner")
     

   • 将处理后的DataFrame覆盖写入目标路径（注意用原子替换方式避免数据丢失）。

方案3：增量加载+替换重复记录（适合小数据量） #

如果数据量不大，可分两步操作：

• 先用Copy Activity将增量数据写入ADLS的临时文件夹。
• 再用Data Flow读取临时数据和目标数据，过滤掉目标数据中与临时数据ID重复的记录，将过滤后的现有数据和临时数据合并，写入新路径后替换原目标路径。

注意：Parquet是列存格式，本身不支持行级修改，所有Upsert逻辑本质都是重写数据，操作时要通过原子替换、版本化存储（如按日期分区）确保数据一致性。

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