Spark Delta流处理出现非真实重复数据问题排查求助

我有一张名为main_table的Delta表，表中记录以unique_id为唯一键，无重复数据。使用以下Spark Streaming配置读取该表的变更流：

spark.readStream.format("delta").option("readChangeFeed", "true").option("maxFilesPerTrigger", 1000).option("maxBytesPerTrigger", 147483648)
                          .option("failOnDataLoss", "true")
                          .load("main_table")
                          .filter(expr("_change_type not in ('delete', 'update_preimage')"))
                          .writeStream
                          .queryName('some_query_name')
                          .foreachBatch(main_logic)
                          .option("checkpointLocation", 'some_path_for_checkpoint')
                          .option("mergeSchema", "true")
                          .trigger(processingTime='1 seconds')
                          .start()

流处理逻辑为对数据进行计算转换（如单位转换）后，通过merge写入目标表。但流处理中部分记录出现重复，而用spark.read.format('delta')批读取该表时，对应unique_id仅返回一条正确记录。
重复记录的update_date与原表不同，且流中读取的是表的历史版本（如原表版本为60，流中读取56、58版本），导致merge时触发报错[DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE]。
目前暂不想用窗口去重，希望找到问题根因，请问是否有类似案例或相关排查文档可参考？

1. Checkpoint状态异常导致版本回溯 #

• 若流任务曾意外重启，checkpoint中记录的处理偏移量可能与Delta表实际变更进度不匹配，导致任务重启后从断点重复拉取已处理过的历史版本变更记录。
• 若未显式指定startingVersion或startingTimestamp参数，首次启动会从表的最新版本开始，但如果checkpoint目录存在残留的异常状态，也会触发版本回溯读取。

2. 变更流过滤逻辑未处理多版本更新 #

• 你的过滤条件仅排除了delete和update_preimage，但同一unique_id的多次连续更新会生成多条update_postimage记录，变更流会完整保留这些历史变更事件；而批读取直接读取表的最新快照，因此不会出现重复。若流处理未基于unique_id+版本号做幂等过滤，就会把多条历史更新记录传入merge逻辑。

3. Trigger配置与文件处理能力不匹配 #

• 你设置了1秒的触发间隔，同时配置了maxFilesPerTrigger和maxBytesPerTrigger，当Delta表的变更文件生成速度快于流处理速度时，可能出现跨批次的文件重叠读取，或同一变更事件被拆分到多个批次重复处理。
• 过小的触发间隔可能导致任务来不及完成checkpoint的持久化，进而在重启时重复读取已处理的文件。

4. Delta表版本清理与流处理的兼容性问题 #

• Delta表默认开启版本自动清理（delta.logRetentionDuration为30天），若流处理速度跟不上版本清理速度，可能出现流尝试读取已被清理的旧版本（触发failOnDataLoss报错）；若清理逻辑异常，也可能残留部分历史版本，导致重复读取。

典型相似场景 #

• 多个流任务共享同一checkpointLocation：会导致偏移量记录混乱，出现重复读取历史版本的情况。
• Merge逻辑未处理多源记录：当流中传入同一unique_id的多条历史更新记录时，Delta Merge会因多条源记录匹配同一目标行触发DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE报错，这是Merge的预期行为——要求源数据中每个匹配键对应唯一记录。

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