从 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

关于这 2 行：

    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