使用Synapse映射数据流批量处理ADLS文件时如何避免主键重复？

你的问题根源在于：批量读取所有增量文件时，同一主键的多行数据会被同时送入UPSERT逻辑，但目标表为空时，所有行都满足插入条件，导致重复插入。要在单个数据流内解决，关键是在写入目标表前先对数据去重，保留每个主键的最新版本，具体步骤如下：

步骤1：源端保留文件名元数据 #

在ADLS源的「源选项」里勾选「添加文件名列」，自动生成包含文件名的列（默认命名为FileName），这个字段用来判断数据的时间先后顺序。

步骤2：派生列提取时间排序依据 #

添加「派生列」转换，从文件名里提取能代表时间顺序的字段。比如你的文件名是daily_data_20240520.csv这种格式，用表达式提取日期部分：

toDate(substring(FileName, 12, 8), 'yyyyMMdd')

将这个派生列命名为FileDate；如果文件名带时间戳，可对应调整substring的参数适配你的命名规则。

步骤3：用窗口转换标记最新行 #

添加「窗口」转换：

• 在「分组依据」里选择你的主键列（比如user_id）
• 在「窗口列」中新增行号字段RowNum，设置：
  • 排序规则：按FileDate（或直接按FileName）降序排列，确保最新文件的行排在每组首位
  • 表达式：rowNumber()

步骤4：过滤只保留最新行 #

添加「过滤」转换，设置过滤条件为：

RowNum == 1

此步骤会让每个主键仅保留来自最新文件的那一行数据，彻底消除数据流内的主键重复。

步骤5：目标表UPSERT配置 #

将过滤后的数据流连接到Synapse目标表，配置UPSERT规则：

• 更新方法选择「Upsert」
• 键列设置为你的主键列
• 勾选「允许插入」和「允许更新」

方案优势 #

• 全程在单个映射数据流内完成，无需使用Foreach activity，避免了集群重复启动的耗时与额外成本
• 无论目标表是空表还是已有数据，写入前都确保了每个主键仅存在一行最新数据，UPSERT逻辑可正常执行：空表时插入唯一行，已有数据时更新对应主键的记录
• 目标表的非强制主键不影响此逻辑，因为重复数据已在数据流内部提前清理

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