pyspark - 从 hdfs 目录迭代 pyspark 中的文件

Question

我在 hdfs 目录中有文件列表，我想从 hdfs 目录遍历 pyspark 中的文件，并将每个文件存储在一个变量中，并使用该变量进行进一步处理。我在下面收到错误..

py4j.protocol.Py4JError: An error occurred while calling z:org.apache.spark.api.python.PythonUtils.toSeq. Trace: 
py4j.Py4JException: Method toSeq([class org.apache.hadoop.fs.Path]) does not exist

InputDir = "/Data/Ready/ARRAY_COUNTERS" # 输入 hdfs 目录。

hadoop = sc._jvm.org.apache.hadoop
fs = hadoop.fs.FileSystem
conf = hadoop.conf.Configuration()
path = hadoop.fs.Path(InputDir)

for f in fs.get(conf).listStatus(path):
    Filename =  f.getPath()

df = spark.read.csv(Filename,header=True)
#I am getting above error in while reading this file.

Answer 1

关于这两行:

    Filename =  f.getPath()

df = spark.read.csv(Filename,header=True)

getPath()不是字符串。此外 - f 也可以是目录，因此要确保您没有尝试加载目录，您可以在 f.isFile() 上添加验证:

if(f.isFile()):
    Filename =  f.getPath()
    df = spark.read.csv(str(Filename),header=True)

现在对我有用的替代方案是:

if(f.isFile()):
    Filename =  f.getPath()
    df = sc.textFile(str(Filename), 500).map(lambda x: x.split(", ")) #or any other spearator, returns RDD
    headers=df.first() # to infer schema - you can then convert it to pyspark dataframe with specific column types