在Databricks中用PySpark覆盖Azure Data Lake Parquet文件遇问题求助

你当前的代码是写入Delta Lake格式，而非直接写入Parquet文件。Delta Lake通过事务日志管理底层的Parquet文件，replaceWhere参数实现的是逻辑覆盖——它会在事务日志中标记符合条件的旧数据为无效，同时写入新的Parquet文件，并不会立即物理删除旧文件，这就是你看到ADLS中不断生成新Parquet文件的原因。

针对你的需求，可通过以下配置调整：

1. 物理清理旧文件（VACUUM） #

如果需要彻底删除被标记为无效的旧Parquet文件，可执行Delta表的VACUUM操作：

from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/mnt/test/new/")
# 保留最近7天的文件，可根据需求调整天数
delta_table.vacuum(7)

注意：VACUUM默认会阻止删除7天内的文件，避免影响未完成的事务，如需修改这个限制，需先设置spark.databricks.delta.retentionDurationCheck.enabled = false（不建议在生产环境随意修改）。

2. 控制新文件数量（避免小文件） #

如果每次运行生成过多小文件，可通过以下参数限制文件大小：

• 在写入前设置Spark配置：

# 每个文件最多存储100000条记录，可根据数据量调整
spark.conf.set("spark.sql.files.maxRecordsPerFile", "100000")
# 写入前重新分区，减少文件数
df = df.repartition("year_id", "month_id", "time_key")

3. 确认replaceWhere的有效性 #

确保replaceWhere的条件完全匹配分区列逻辑：你的代码中time_key是分区列，条件time_key = 20231020是正确的，但如果分区列是组合的（year_id/month_id/time_key），也可以把条件写得更精确，比如：

.option("replaceWhere", "year_id=2023 AND month_id=10 AND time_key=20231020")

这能确保只覆盖目标分区，避免误操作其他分区的数据。

额外说明 #

如果你原本的需求是直接写入Parquet文件而非Delta表，需要修改代码的格式为format("parquet")，同时结合mode("overwrite")和partitionBy，但这种方式没有事务管理，一旦写入错误无法回滚：

(
    df.write
    .format("parquet")
    .mode("overwrite")
    .partitionBy("year_id","month_id","time_key")
    .option("replaceWhere", "time_key = 20231020")
    .save("/mnt/test/new/")
)

但这种直接写Parquet的replaceWhere仅在Spark 3.0+支持，且需要确保分区列的条件准确。

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