DeltaLake（Python）合并/Upsert操作后重复插入问题排查求助

我现在遇到了DeltaLake合并操作的重复数据问题，希望能得到大家的帮助。具体情况如下：

环境信息 #

• 后端存储：Azure Blob Storage
• 使用的库版本：
  • deltalake 1.1.4（Python）
  • Polars 1.31.0（数据源为LazyFrame/DataFrame）
• 目标：实现幂等Upsert——重复执行相同输入不会产生新行

Delta表结构 #

Schema(
    [
        Field("area_type_code", PrimitiveType("string"), nullable=True),
        Field("map_code", PrimitiveType("string"), nullable=True),
        Field("fuel", PrimitiveType("string"), nullable=True),
        Field("datetime", PrimitiveType("timestamp_ntz"), nullable=True),
        Field("period_name", PrimitiveType("string"), nullable=True),
        Field("period_granularity", PrimitiveType("string"), nullable=True),
        Field("power", PrimitiveType("double"), nullable=True),
        Field("energy", PrimitiveType("double"), nullable=True)
    ]
)

Upsert实现逻辑 #

我将数据源拆分成不同大小的块（试过2M和10M行），每处理一个块前会重新加载Delta表，确保之前块的插入/更新操作可见。然后执行合并操作：

merge_results = delta_table.merge(
    source=df_chunk,
    predicate=merge_predicate,
    source_alias='source',
    target_alias='target',
    writer_properties=writer_properties,
    streamed_exec=True,
).when_matched_update(
    predicate=match_predicate,
    updates=update_mapping
).when_not_matched_insert(
    updates=insert_mapping
).execute()

谓词与映射规则 #

• 合并谓词（Merge Predicate）：
  最初使用的规则：

  target.area_type_code = source.area_type_code 
  AND target.map_code = source.map_code 
  AND target.fuel = source.fuel 
  AND target.datetime = source.datetime 
  AND target.period_granularity = source.period_granularity
  

  后续尝试基于当前块的唯一值添加分区过滤：

  AND target.period_granularity IN ('hourly', 'daily') 
  AND target.area_type_code IN ('BZN')
  

• 匹配谓词（Match Predicate）：

  target.power != source.power OR target.energy != source.energy
  

• 更新映射：

  {'power': 'source.power', 'energy': 'source.energy'}
  

• 插入映射：

  {
      'period_name': 'source.period_name',
      'period_granularity': 'source.period_granularity',
      'area_type_code': 'source.area_type_code',
      'energy': 'source.energy',
      'power': 'source.power',
      'map_code': 'source.map_code',
      'datetime': 'source.datetime',
      'fuel': 'source.fuel'
  }
  

我的理解是：合并谓词用来判断源数据中的记录是否已存在于目标表；匹配谓词用来决定已存在的记录是否需要更新；映射规则则指定源数据的哪些值要写入目标表的对应列。

问题现象 #

• 第一次执行Upsert：Delta表成功创建，总行数为10,240,472，和输入DataFrame的行数完全一致。
• 重复执行相同输入：merge操作返回的结果显示有新的插入和更新（如下示例），加载Delta表后能看到重复行，而且我对比过重复行的所有列值，完全没有差异。
  执行结果示例：

  {
      'num_source_rows': 240472,
      'num_target_rows_inserted': 29782,
      'num_target_rows_updated': 4429,
      'num_target_rows_deleted': 0,
      'num_target_rows_copied': 471766,
      'num_output_rows': 505977,
      'num_target_files_scanned': 21,
      'num_target_files_skipped_during_scan': 0,
      'num_target_files_added': 20,
      'num_target_files_removed': 18
  }
  

• 我已经确认合并谓词中用到的所有列都没有NULL或NAN值；试过把datetime替换成period_name（字符串格式的时间），但还是会出现重复插入的问题。

疑问 #

想请教大家几个问题：

• 我的merge/match逻辑是不是哪里有问题，导致无法实现幂等Upsert？
• 浮点数（double类型的power/energy）比较有没有已知的边缘情况，会导致判断不等或者跨批次重复插入？
• 针对timestamp_ntz类型，有没有推荐的确定性匹配方法（比如精度归一化）？
• 分块处理时每次重新加载表，有哪些最佳实践可以避免重复（比如事务策略、分区谓词、写入属性）？
• 是否建议先执行删除再合并，或者有更高效的保障方式？

感谢大家的任何建议和排查方向，我可以提供更多细节。

内容来源于stack exchange