如何让Databricks Autoloader重新处理被覆盖的已摄入文件？

1. 开启文件校验和检测（推荐） #

Autoloader支持通过校验和追踪文件内容变更，而非仅依赖文件名。开启后，当文件被覆盖且内容变化时，Autoloader会自动重新处理该文件。

修改你的PySpark代码，在readStream配置中添加cloudFiles.checksum选项：

# Configure Auto Loader to ingest csv data to a Delta table
query = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", source_format)
    .option("cloudFiles.schemaLocation", checkpoint_directory)
    .option("cloudFiles.checksum", "true")  # 开启校验和检测
    .option("header", "true")
    .option("delimiter", ";")
    .option("skipRows", 7)
    .option("pathGlobFilter", "AP_SAPEX_KPI_001 - Posted Invoices in *.CSV")
    .load(data_source)
    .select(
        "*", 
        current_timestamp().alias("_JOB_UPDATED_TIME"), 
        input_file_name().alias("_JOB_SOURCE_FILE"), 
        col("_metadata.file_modification_time").alias("_MODIFICATION_TIME")
    )
    .writeStream
    .option("checkpointLocation", checkpoint_directory)
    .option("mergeSchema", "true")
    .trigger(availableNow=True)
    .toTable(table_name)
)

原理：该选项会计算每个文件的MD5校验和，并将其存储在checkpoint中。当文件被覆盖后，新的校验和与存储值不一致，Autoloader会判定这是一个需要重新处理的“新”文件。

2. 结合时间过滤与部分重置Checkpoint（备选） #

如果无法使用校验和，可以通过更新modifiedAfter时间戳并重置checkpoint的偏移量目录，迫使Autoloader重新扫描指定时间后的文件（包括已被覆盖的文件）：

• 每次运行前，将modifiedAfter设置为上次任务的结束时间（例如前一天的运行时间）
• 删除checkpoint目录下的offsets子目录（保留schema目录避免重新推断Schema）：

  dbutils.fs.rm(checkpoint_directory + "/offsets", recurse=True)
  

• 运行原任务代码

注意：此方法会重新处理所有符合modifiedAfter条件的文件，可能导致重复数据。建议配合Delta Lake的Merge操作，用最新的文件数据替换旧数据，或者在下游表中基于_MODIFICATION_TIME保留最新版本。

3. 修改文件命名规则（最佳实践） #

最彻底的解决方式是调整上游文件命名规则，避免同名覆盖。例如在文件名中添加日期维度，如AP_SAPEX_KPI_001 - Posted Invoices in 20240915.CSV（包含年月日）。

这样每个文件都是唯一的，Autoloader会自动捕获新文件，同时保留所有历史版本，便于数据回溯和问题排查。

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