Azure Synapse中PySpark读写Delta Lake时的数据重复问题

针对你遇到的合并bronze和silver数据后去重无效、银层仍存在重复Id的问题，以下是几个实用的排查和解决方向：

1. 检查Id列的隐性差异 #

看起来相同的Id可能存在大小写不一致、前后空格、全角/半角字符等隐性差异，导致dropDuplicates无法识别为重复值。比如"123"和" 123"、"ID123"和"id123"会被判定为不同值。

解决办法： #

先对Id列做标准化清洗，确保格式统一：

from pyspark.sql.functions import trim, col

# 对bronze和silver的Id列统一处理：去除前后空格、转成字符串类型、统一大小写
bronze_df = bronze_df.withColumn("Id", trim(col("Id")).cast("string").upper())
silver_df = silver_df.withColumn("Id", trim(col("Id")).cast("string").upper())

# 再执行合并和去重
deduplicated_df = bronze_df.unionByName(silver_df).dropDuplicates(["Id"])

2. 替换union为unionByName #

PySpark的union是按字段位置合并，而非按列名匹配。如果bronze和silver表的字段顺序不一致，会导致Id列被错位赋值，去重逻辑完全失效。

解决办法： #

使用unionByName按列名合并，确保字段对齐：

# 用unionByName替代union，避免字段错位
deduplicated_df = bronze_df.unionByName(silver_df).dropDuplicates(["Id"])

3. 改用窗口函数实现精准去重（保留最新记录） #

dropDuplicates默认保留数据集里第一条出现的记录，依赖合并顺序，无法确保保留最新的业务数据（比如Salesforce的LastModifiedDate更新记录）。如果重复是因为新数据未覆盖旧数据，这种方法更可靠。

解决办法： #

通过窗口函数按Id分组，按更新时间降序排序，只保留最新的一条记录：

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, col

# 定义窗口：按Id分组，按LastModifiedDate倒序排列（确保最新记录排在最前）
window_spec = Window.partitionBy("Id").orderBy(col("LastModifiedDate").desc())

# 合并数据后添加行号，过滤出每个Id的第一条（最新）记录
deduplicated_df = bronze_df.unionByName(silver_df) \
    .withColumn("row_num", row_number().over(window_spec)) \
    .filter(col("row_num") == 1) \
    .drop("row_num")

4. 排查Delta Lake的写入与元数据问题 #

如果以上操作后仍有重复，可能是Delta Lake的旧版本数据未被彻底覆盖，或者表元数据存在异常：

解决办法： #

• 执行Delta表优化与清理，删除旧版本数据：

from delta.tables import DeltaTable

delta_silver_table = DeltaTable.forPath(spark, prod_user_silver_path)
# 优化表结构
delta_silver_table.optimize().execute()
# 清理7天前的旧版本（如果需要立即清理可设为0小时，注意：会无法恢复旧数据）
delta_silver_table.vacuum(24*7)

• 确认写入模式：确保mode("overwrite")是覆盖整个表，而非仅部分分区（如果是分区表，可添加.option("overwriteSchema", "true")确保 schema 同步）。

5. 提前对bronze层数据去重 #

如果bronze层本身就存在重复Id，会导致合并后去重压力大，甚至可能因为数据量问题出现异常。

解决办法： #

先对bronze层单独去重，再与silver层合并：

# 先清理bronze层内部的重复Id
bronze_deduped_df = bronze_df.dropDuplicates(["Id"])
# 再合并silver层并去重
deduplicated_df = bronze_deduped_df.unionByName(silver_df).dropDuplicates(["Id"])

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