Databricks Autoloader在ADLS Gen2多文件夹下的数据源路径配置问题

一、多文件夹场景下的Autoloader工作逻辑 #

Autoloader核心是增量监听存储路径下的新增文件，不管是单文件夹还是多子文件夹结构，都能自动发现新写入的文件。在你的多表场景中，根目录下每个子文件夹对应一张表，Autoloader既可以监听根目录，通过提取文件路径信息关联对应表；也可以单独监听某个子文件夹，只处理单表增量数据。

它会通过checkpoint机制记录已处理文件，不会重复处理，完美适配你每15分钟新增数据的场景。

二、data_source路径配置方案 #

假设你的ADLS Gen2存储结构如下：

abfss://<容器名>@<存储账户名>.dfs.core.windows.net/根目录/
├─ table_a/
│  ├─ data_20240501_0015.csv
│  └─ ...
├─ table_b/
│  ├─ data_20240501_0015.csv
│  └─ ...
└─ ...

方案1：批量监听所有表的文件夹 #

如果需要一次性处理所有表的增量数据，data_source直接填根目录路径，再通过文件路径提取对应表名：

from pyspark.sql.functions import input_file_name, regexp_extract

# 读取所有子文件夹的新增CSV数据
raw_df = (spark.readStream
          .format("cloudFiles")
          .option("cloudFiles.format", "csv")
          # 存储自动推断的schema，支持后续schema演化
          .option("cloudFiles.schemaLocation", "/dbfs/checkpoints/autoloader/global_schema")
          # 限制每次触发处理的文件数，按需调整
          .option("cloudFiles.maxFilesPerTrigger", 100)
          # 填写根目录路径
          .load("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/根目录/")
          # 从文件路径提取表名，正则规则根据实际路径调整
          .withColumn("table_name", regexp_extract(input_file_name(), r"根目录\/(\w+)\/", 1))
)

# 按表名分流写入对应Delta表示例
table_a_df = raw_df.filter(raw_df.table_name == "table_a")
query_a = (table_a_df.writeStream
           .format("delta")
           .option("checkpointLocation", "/dbfs/checkpoints/autoloader/table_a_checkpoint")
           # 匹配15分钟的增量数据加载频率
           .trigger(processingTime='15 minutes')
           .table("你的数据库名.table_a")
)

方案2：单独监听单表文件夹 #

如果只需要处理某一张表（比如table_a），data_source直接填该表对应的子文件夹路径：

# 仅监听table_a文件夹的新增数据
table_a_df = (spark.readStream
              .format("cloudFiles")
              .option("cloudFiles.format", "csv")
              .option("cloudFiles.schemaLocation", "/dbfs/checkpoints/autoloader/table_a_schema")
              # 填写单表的子文件夹路径
              .load("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/根目录/table_a/")
)

# 写入Delta表
query_a = (table_a_df.writeStream
           .format("delta")
           .option("checkpointLocation", "/dbfs/checkpoints/autoloader/table_a_checkpoint")
           .trigger(processingTime='15 minutes')
           .table("你的数据库名.table_a")
)

三、关键配置注意事项 #

• 权限验证：确保Databricks集群拥有ADLS Gen2存储账户的读写权限（可通过服务主体、SAS令牌或RBAC配置）
• Checkpoint独立性：每个表的流处理任务必须配置独立的checkpointLocation，避免不同任务的状态冲突
• Schema演化支持：如果CSV文件可能新增字段，可开启cloudFiles.schemaEvolutionMode选项，比如设置为"addNewColumns"自动适配新字段
• 触发频率：通过trigger(processingTime='15 minutes')让任务每15分钟触发一次，和你的数据加载频率对齐

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