启用文件通知的Autoloader如何排除最新加载文件夹？

方案1：数据分层归档（推荐，从根源避免问题） #

把数据湖分成临时区和归档区两个层级：

• 写入端先将实时更新的文件夹存入/raw/temp/目录，这个目录不被Autoloader监听。
• 当确认该文件夹停止写入（比如写入端完成数据推送、或监控到文件夹内文件10分钟无更新），再将整个文件夹移动到/raw/archive/目录。
• 配置Autoloader只监听/raw/archive/**路径，这样只会读取已经稳定、不再更新的文件夹，彻底避免读取正在写入的CSV文件导致的失败。

示例配置（Scala）：

val df = spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.useNotifications", "true")
  .option("cloudFiles.checkpointLocation", "/path/to/checkpoint")
  .load("/raw/archive/")

这种方案的优势：

• 完全兼容文件通知机制，Autoloader可以自动捕获归档区的新文件夹，无需手动维护路径列表。
• 从根源隔离了不稳定的实时数据和可读取的归档数据，避免读取失败。
• 确保所有数据最终都会被加载（只要写入端完成归档操作），不会丢失数据。

方案2：动态过滤最新文件夹（无需修改写入流程） #

如果无法修改写入端的路径规则，可以通过动态文件过滤让Autoloader自动排除最新的文件夹：

1. 在流作业启动时（或每个微批开始前），查询数据湖中的所有文件夹，提取时间戳并找出最新的那个。
2. 定义一个文件过滤函数，排除该最新文件夹下的所有文件。
3. 将过滤函数传给Autoloader，实现实时排除。

示例配置（Scala）：

// 1. 获取数据湖根目录下的所有文件夹，提取最新的时间戳文件夹
val rootPath = "/raw/"
val folders = spark.sql(s"SELECT DISTINCT split_part(path, '/', -2) AS folder_name FROM fs.glob('${rootPath}*')")
val latestFolder = folders.orderBy(desc("folder_name")).first().getString(0)

// 2. 定义过滤函数：排除最新文件夹下的文件
val excludeLatest = (filePath: String) => !filePath.contains(latestFolder)

// 3. 配置Autoloader
val df = spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.useNotifications", "true")
  .option("cloudFiles.checkpointLocation", "/path/to/checkpoint")
  .option("cloudFiles.fileFilter", excludeLatest)
  .load(rootPath)

注意事项： #

• 如果是持续运行的流作业，需要定期刷新latestFolder的值（比如每小时重启一次作业，或用广播变量结合触发器更新），否则当新的文件夹生成后，旧的最新文件夹会一直被排除，导致数据遗漏。
• 当最新文件夹停止更新后，在下一次刷新过滤规则时，它会被纳入读取范围，Autoloader会自动处理其中的所有文件（依赖checkpoint记录，不会重复读取已处理的文件）。

避免重复读取的关键 #

不要手动构造路径列表（比如{2023-10-09T12.31.31Z,2023-10-09T14.02.15Z}），这种方式会破坏Autoloader的文件通知和checkpoint机制：

• 文件通知是监听整个根路径的新增文件/文件夹，手动指定路径会导致通知逻辑失效，无法自动捕获新文件夹。
• 重复传入已读取的文件夹，Autoloader虽然会基于checkpoint跳过已处理文件，但会增加不必要的元数据扫描开销。

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