PySpark中带分区的Delta表覆盖写入未按预期生效

核心原因 #

你遇到的Delta Lake写入后旧数据残留问题，主要和Delta Lake的overwrite模式行为、事务完整性，以及repartition的使用细节有关，常见诱因包括：

1. 未提交的事务残留
   如果第一个笔记本的写入操作因中断、资源不足等原因未完全提交，Delta的事务日志会保留旧文件的有效引用，导致查询时仍能读取到旧数据。手动删除目录相当于彻底清除了所有旧文件和日志，自然解决了问题。

2. repartition导致的文件替换不彻底
   使用repartition('id')重新分区后，生成的文件数量、命名规则和之前的写入结果可能不同。若写入过程中Spark未完全替换旧文件（比如部分文件写入失败但事务未回滚），Delta日志不会标记这些旧文件为无效，最终查询时会混合新旧数据。

3. Delta版本保留的极端情况
   Delta Lake默认保留7天历史版本，若查询时因意外未指向最新版本，可能读取到旧数据（此情况概率较低，默认查询会指向最新版本）。

解决方案 #

针对这些问题，你可以尝试以下步骤：

1. 强制全量覆盖数据 #

在写入时添加replaceWhere参数，强制Delta Lake覆盖所有数据，避免因分区或文件匹配问题导致的旧数据残留：

df1.repartition('id').write.format("delta")
  .option("overwriteSchema", "True")
  .option("replaceWhere", "1=1")  # 强制覆盖全表
  .mode("overwrite")
  .save('/transformed/output2')

2. 检查并修复Delta表状态 #

通过Delta元数据命令查看表的历史事务，确认是否存在未提交或失败的写入，必要时修复表：

# 查看表的历史版本与事务状态
spark.sql("DESCRIBE HISTORY delta.`/transformed/output2`").show()

# 修复存在异常的Delta表
spark.sql("REPAIR TABLE delta.`/transformed/output2`")

3. 验证写入前的数据集 #

确认df1确实不包含已删除的目标行，排除上游数据读取错误导致的旧数据重复写入：

# 替换为被删除行的id值，验证数据集是否已移除该行
df1.filter(df1.id == "目标id值").show()

4. 写入前主动清理目录（可选） #

如果上述方法无效，可在写入前主动清理目标目录，确保从空状态开始写入：

dbutils.fs.rm('/transformed/output2', recurse=True)

# 执行写入操作
df1.repartition('id').write.format("delta")
  .option("overwriteSchema", "True")
  .mode("overwrite")
  .save('/transformed/output2')

5. 排查并发写入 #

确认没有其他进程或笔记本同时写入/transformed/output2目录，并发写入可能导致事务冲突，破坏Delta表的一致性。

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