如何过滤数据块自动加载程序流中的文件

Question

我想使用数据砖自动加载器设置S3流。我已经设法设置了流，但我的S3存储桶包含不同类型的JSON文件。我想过滤掉它们，最好是在流本身中，而不是使用过滤操作。

根据文档，我应该能够使用全局模式进行过滤。但是，我似乎无法让它工作，因为它无论如何都会加载所有内容。

这是我有的

df = (
  spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaInference.samleSize.numFiles", 1000)
  .option("cloudFiles.schemaLocation", "dbfs:/auto-loader/schemas/")
  .option("includeExistingFiles", "true")
  .option("multiLine", "true")
  .option("inferSchema", "true")
#   .option("cloudFiles.schemaHints", schemaHints)
#  .load("s3://<BUCKET>/qualifier/**/*_INPUT")
  .load("s3://<BUCKET>/qualifier")
  .withColumn("filePath", F.input_file_name())
  .withColumn("date_ingested", F.current_timestamp())
)

我的文件有一个结构为限定符/version/YYYY-MM/DD/的键

这似乎加载了所有内容：. load（"s3：//

是我的glob模式不正确，还是我遗漏了什么？




Answer 1

从留档来看，你似乎可以同时使用加载和path Globfilter选项来实现你所需要的。你在这里尝试过这样的方法吗？

df = (
  spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaInference.samleSize.numFiles", 1000)
  .option("cloudFiles.schemaLocation", "dbfs:/auto-loader/schemas/")
  .option("includeExistingFiles", "true")
  .option("multiLine", "true")
  .option("inferSchema", "true")
  .option("pathGlobfilter", "*_INPUT.json") 
  .load("s3://<BUCKET>/qualifier/")
  .withColumn("filePath", F.input_file_name())
  .withColumn("date_ingested", F.current_timestamp())
)