AutoLoader读取ADLS写入中文件的数据丢失风险及Trigger.AvailableNow咨询

问题背景 #

使用Databricks AutoLoader将ADLS上的Parquet文件摄入Delta表，源系统独立写入文件，存在启动readStream时目标文件仍在写入的场景。当前配置为目录列表模式（cloudFiles.usenotifications=False）+trigger(once=True)，核心代码如下：

cloudFile = {
         "cloudFiles.usenotifications": False
        , "cloudFiles.format": self.source_format
        , "cloudFiles.schemaLocation": self.al_output_path_schema
        , "cloudFiles.maxBytesPerTrigger": '1g'
        , "cloudFiles.inferColumnTypes": True
        , "cloudFiles.validateOptions": False
    }

df = spark.readStream\
  .format("cloudFiles")\
  .options(**cloudFile)\
  .option("readerCaseSensitive", "false")\
  .option("ignoreLeadingWhiteSpace", "true")\
  .option("ignoreTrailingWhiteSpace", "true")\
  .option("recursiveFileLookup", "true")\
  .load(input_path)

df.writeStream\
    .foreachBatch(lambda df, epoch_id: self._al_write_data(df))\
    .option( "checkpointLocation", self.al_output_path_checkpoint )\
    .option( "mergeSchema", "true")\
    .queryName(f'_al_stream_{self.target_schema}_{self.target_name}')\
    .trigger(once=True)\
    .start()

需明确两个核心问题：

1. 当前配置下AutoLoader的行为，是否存在数据丢失风险？
2. Trigger.AvailableNow能否解决该问题？

一、当前配置（目录列表模式+trigger(once=True)）的行为与风险 #

1. AutoLoader的处理逻辑 #

目录列表模式下，AutoLoader通过扫描目录元数据（文件大小、修改时间）判断文件是否就绪：

• 扫描时如果检测到文件大小仍在变化、修改时间频繁更新，会标记为「未就绪」，暂不读取。
• 但trigger(once=True)模式下，流任务仅执行一次目录扫描+数据处理就停止，不会再次扫描目录。

2. 数据丢失风险 #

• 启动时未完成写入的文件，会被标记为未就绪，但任务停止后不会再重新检查这些文件的状态，导致这类文件被永久跳过，存在明确的数据丢失风险。
• 如果AutoLoader误判文件已就绪（比如写入过程中短暂停顿，大小临时稳定），读取不完整的Parquet文件会直接抛出解析错误，或摄入损坏的数据。

二、Trigger.AvailableNow的作用与效果 #

Trigger.AvailableNow（需DBR 11.3+/Spark 3.3+）是为批量处理流数据设计的触发模式，核心特性是：

• 任务启动后会持续扫描目录，直到处理完所有已就绪的文件后自动停止，而非仅扫描一次。
• 对于第一次扫描时标记为未就绪的文件，会在后续扫描轮次中重新检查状态，当文件写入完成（大小稳定、修改时间不再变化），立即读取处理。

是否能解决问题？ #

是，它能解决当前trigger(once=True)的核心问题：

• 避免了单次扫描导致未完成文件被永久跳过的情况，只要文件在任务运行期间完成写入，就会被捕获处理。
• 即使文件写入持续时间较长，只要任务未停止，就会反复扫描检查，直到无新的就绪文件。

注意事项 #

如果源系统的文件写入完全非原子（无临时文件重命名，持续波动），需依赖AutoLoader的文件稳定性检测机制：默认会通过多次校验文件元数据确认是否就绪，只要文件最终稳定，就会被处理。若任务结束前文件仍未写完，下次启动AvailableNow任务时，会从checkpoint中读取未就绪文件的记录，重新检查处理。

三、额外优化建议 #

1. 强制源系统使用原子写入：先写入临时目录（如_temp/），完成后重命名至目标目录。ADLS的重命名操作是原子性的，AutoLoader只会扫描到已完成的文件，从根源避免未就绪文件问题。
2. 开启文件校验：将cloudFiles.validateOptions设为True，AutoLoader会校验Parquet文件的完整性，避免读取损坏数据。
3. 调整扫描间隔（可选）：目录列表模式下可通过cloudFiles.fileNotificationInterval调整扫描间隔，但AvailableNow会自动适配，一般无需手动配置。

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