如何在HDFS上使用SparkStreaming时获取文件名

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from pyspark.sql import SparkSession
import os

# 创建SparkContext对象
sc = SparkContext(appName="SparkStreamingExample")
# 创建StreamingContext对象，设置批处理间隔为5秒
ssc = StreamingContext(sc, 5)

# 创建一个DStream，指定输入源为HDFS目录
dstream = ssc.textFileStream("hdfs://localhost:9000/input")

# 使用transform方法处理每个RDD
transformed_stream = dstream.transform(lambda rdd: 
    rdd.mapPartitionsWithIndex(lambda idx, it: 
        [(os.path.basename(x), x) for x in it]))

# 对每个文件名和数据进行进一步处理或存储
transformed_stream.foreachRDD(lambda rdd: 
    rdd.foreach(lambda x: 
        print("File name: {}, Data: {}".format(x[0], x[1]))))

# 启动StreamingContext
ssc.start()
ssc.awaitTermination()