如何在Databricks Autoloader流中过滤S3存储桶的指定JSON文件

问题原因 #

你遇到的规则不生效问题核心是两个原因：

1. Auto Loader 对load()方法中直接写的glob路径兼容性有限，部分Databricks Runtime版本不会自动递归解析多层目录的通配符规则，官方更推荐用专用参数配置文件过滤规则
2. 你现有代码中存在参数拼写错误，可能间接影响配置加载逻辑

修正方案 #

直接使用cloudFiles.pathGlobPattern参数配置过滤规则，修正后的代码如下：

df = (
  spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.inferColumnTypes", "true")
  # 修正拼写错误：samleSize -> sampleSize
  .option("cloudFiles.schemaInference.sampleSize.numFiles", 1000)
  .option("cloudFiles.schemaLocation", "dbfs:/auto-loader/schemas/")
  .option("includeExistingFiles", "true")
  .option("multiLine", "true")
  .option("inferSchema", "true")
  # 新增glob过滤规则，匹配所有文件名以_INPUT.json结尾的文件
  .option("cloudFiles.pathGlobPattern", "**/*_INPUT.json")
#   .option("cloudFiles.schemaHints", schemaHints)
  .load("s3://<BUCKET>/qualifier")
  .withColumn("filePath", F.input_file_name())
  .withColumn("date_ingested", F.current_timestamp())
)

注意事项 #

• 如果你的流任务之前已经启动过，schemaLocation路径下已经记录了历史加载的文件索引，之前已经加载过的非INPUT文件不会被自动回溯剔除。如果需要全量重新按规则加载，需要先清空dbfs:/auto-loader/schemas/目录，再启动任务。
• 该配置的过滤逻辑是在文件加载阶段执行，不会走到后续计算层的filter，符合你的需求。

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