如果存储类型与sql类型兼容，并且使用spark2.0，那么就非常简单了。导入 input_file_name :

from pyspark.sql.functions import input_file_name

读取文件并转换为 DataFrame :

df = sc.sequenceFile("/tmp/foo/").toDF()

添加文件名：

df.withColumn("input", input_file_name())

如果此解决方案不适用于您的情况，那么通用的解决方案是直接列出文件（对于您可以使用的HDF） hdfs3 图书馆）：

files = ...

逐个读取添加文件名：

def read(f):
    """Just to avoid problems with late binding"""
    return sc.sequenceFile(f).map(lambda x: (f, x))

rdds = [read(f) for f in files]

和工会：

sc.union(rdds)