PySpark Streaming使用通配符无法读取子文件夹文件求助

问题描述 #

使用PySpark Streaming读取S3文件时，路径设为directory="s3://bucket/folder/2022/10/18/4/*"这种具体层级加单通配符的格式，代码可正常运行；但改用directory="s3://bucket/folder/*/*/*/*/*"这类多层通配符时，无法读取子文件夹中的文件。官方说明PySpark Streaming支持POSIX glob模式，需解决该问题。

原代码如下：

from pyspark.sql.functions import from_json
from pyspark.streaming import StreamingContext

ssc = StreamingContext(sc, 30)

directory = "s3://bucket/folder/*/*/*/*/*"
stream_data = ssc.textFileStream(directory)

def readMyStream(rdd):
  if not rdd.isEmpty():
    df = spark.read.option("multiline","true").json(rdd)
    print('Started the Process')
    print('Selection of Columns')
    df = df.select("c1","c2","c3","c4","c5")
    df.show()
    
stream_data.foreachRDD(lambda rdd: readMyStream(rdd)) 

ssc.start()
ssc.awaitTermination()

解决方案 #

方法1：改用Spark结构化流（推荐） #

Spark Streaming（DStream API）属于旧版流式处理API，对多层通配符递归读取支持有限。推荐使用结构化流（Structured Streaming），它原生支持递归文件查找和灵活的通配符配置：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("S3RecursiveStream") \
    .getOrCreate()

# 配置流式读取，开启递归查找
stream_df = spark.readStream \
    .option("multiline", "true") \
    .option("recursiveFileLookup", "true") \
    .json("s3://bucket/folder/*")  # 顶层目录加通配符即可覆盖所有子层级

# 选择目标列并输出到控制台（可替换为其他输出Sink）
query = stream_df.select(col("c1"), col("c2"), col("c3"), col("c4"), col("c5")) \
    .writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

query.awaitTermination()

方法2：调整DStream配置（兼容旧API） #

若必须使用DStream API，可通过Hadoop配置开启递归文件查找，同时简化路径写法：

from pyspark.sql.functions import from_json
from pyspark.streaming import StreamingContext

# 开启Hadoop递归文件查找配置
sc._jsc.hadoopConfiguration().set("mapreduce.input.fileinputformat.input.dir.recursive", "true")

ssc = StreamingContext(sc, 30)

# 仅指定顶层目录，依赖递归配置读取所有子目录文件
directory = "s3://bucket/folder/"
stream_data = ssc.textFileStream(directory)

def readMyStream(rdd):
  if not rdd.isEmpty():
    df = spark.read.option("multiline","true").json(rdd)
    print('Started the Process')
    print('Selection of Columns')
    df = df.select("c1","c2","c3","c4","c5")
    df.show()
    
stream_data.foreachRDD(lambda rdd: readMyStream(rdd)) 

ssc.start()
ssc.awaitTermination()

关键说明 #

• textFileStream的设计逻辑是监控指定目录下的新增文件，多层通配符会被解析为固定层级路径，而非递归遍历所有子目录。
• 结构化流的recursiveFileLookup选项可自动递归遍历所有子目录，配合顶层通配符即可覆盖所有层级的文件。
• 使用DStream时，开启Hadoop递归配置后，只需指定顶层目录，无需多层通配符即可读取所有子目录中的文件。

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