如何用Databricks文件到达触发器追踪嵌套文件夹中的文件位置

核心方案：用Auto Loader实现递归文件检测与元数据捕获 #

Databricks的Auto Loader是处理文件到达场景的最优工具，支持递归扫描嵌套文件夹，同时能轻松捕获文件的路径和名称信息，无需手动遍历目录。

1. 配置递归扫描与文件过滤 #

开启recursiveFileLookup参数让Auto Loader遍历所有子文件夹，再通过pathGlobFilter精准匹配你需要的file.csv文件，避免处理无关文件。

2. 捕获文件路径与文件名 #

使用Spark内置函数input_file_name()获取文件的完整路径，再通过正则表达式提取文件夹路径和文件名：

• folder_path：提取文件所在的父文件夹路径
• file_name：提取文件名（逻辑可适配其他命名规则的文件）

代码示例（流处理/持续监控） #

如果需要持续监控文件夹并实时处理新文件，用流处理模式：

from pyspark.sql.functions import input_file_name, regexp_extract

# 读取嵌套文件夹中的file.csv
stream_df = (spark.readStream
             .format("cloudFiles")
             .option("cloudFiles.format", "csv")
             .option("recursiveFileLookup", "true")  # 开启递归扫描
             .option("pathGlobFilter", "file.csv")  # 仅匹配目标文件
             .load("/dbfs/mnt/source1")  # 替换为你的实际存储路径
             # 添加文件元数据列
             .withColumn("full_file_path", input_file_name())
             .withColumn("folder_path", regexp_extract("full_file_path", "(.*)/", 1))
             .withColumn("file_name", regexp_extract("full_file_path", ".*/(.*)", 1))
            )

# 处理后写入Delta表（可替换为你的业务逻辑）
write_query = (stream_df.writeStream
               .format("delta")
               .option("checkpointLocation", "/dbfs/mnt/checkpoints/file_processing")  # 记录处理状态，避免重复
               .table("processed_files")
               .start())

write_query.awaitTermination()

代码示例（批处理/触发式作业） #

如果需要通过“文件到达”触发器单次处理新文件，用批处理模式：

from pyspark.sql.functions import input_file_name, regexp_extract

# 读取新增的file.csv文件
batch_df = (spark.read
            .format("csv")
            .option("recursiveFileLookup", "true")
            .option("pathGlobFilter", "file.csv")
            .load("/dbfs/mnt/source1")
            .withColumn("full_file_path", input_file_name())
            .withColumn("folder_path", regexp_extract("full_file_path", "(.*)/", 1))
            .withColumn("file_name", regexp_extract("full_file_path", ".*/(.*)", 1))
           )

# 业务处理逻辑（示例：追加写入目标表）
batch_df.write.mode("append").saveAsTable("processed_files")

作业触发器配置 #

1. 在Databricks作业中创建新任务，关联你编写的处理脚本
2. 配置触发器为文件到达，指定根目录为/dbfs/mnt/source1（你的实际存储路径）
3. 设置触发条件：比如检测到至少1个新文件时运行作业
4. 确保作业拥有存储目录的读写权限（如ADLS/S3的IAM权限）

关键注意事项 #

• 避免重复处理：流处理模式下的checkpointLocation会自动记录已处理文件的状态；批处理模式可结合Delta Lake的ACID特性或自定义状态表跟踪已处理路径
• 性能优化：如果嵌套层级极深，可调整cloudFiles.maxFilesPerTrigger控制每次处理的文件数量，避免作业过载
• 适配不同存储：上述代码适用于DBFS、ADLS Gen2、S3等Databricks支持的存储系统，只需替换路径即可

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