python - Hadoop/Spark 读取许多 CSV 文件

Question

我有很多以非常有意义的方式存储的结构化数据，我想以同样有意义的完整和高效的方式处理它们。

+- some-hdfs-path/
  +- level-1_var-01/
  |  +- level-2_var-001.csv
  |  +- ...
  |  +- level-2_var-nnn.csv
  +- level-1_var-02/
  |  +- level-2_other-001.csv
  |  +- ...
  |  +- level-2_other-mmm.csv
  +- ... /
  +- level-1_var-nn/
  |  +- ...

每个文件大约 100MB，大约有 1,000,000 行。每个目录中的文件数量(通常约为 100 个)各不相同，文件名也各不相同。换句话说，我不知道有多少文件或它们叫什么，但我确实需要它们的名称，显然还有它们的内容。

我无法处理从 sc.textFile("/some-hdfs-path/level-1_var-01/*.csv") 和 返回的 RDD >sc.wholeTextFiles("/some-hdfs-path/level-1_var-01")。

总体目标是实际获取 level-1_var/目录中每个文件的第一行和最后一行。合并每个 level-1_var 的结果，然后返回并在 some-other-hdfs-path/level-1-var/中为每个 level-1_var/写出一组全新的文件

我是 Hadoop/Spark 的新手，使用的是 RDD。我读过 documentation对于上述两个函数，但我仍然对如何迭代我返回的 RDD 并进行处理感到困惑。

编辑:文件包含时间序列数据，因此不希望将每个目录中的文件内容串联起来。我愿意将文件内容作为附加列添加到一个巨大的数据框中，而不是作为行。

Answer 1

通过替换您的配置和属性，使用此代码在 pySpark 中读取 CSV。

from pyspark.sql import SparkSession
from pyspark.sql import Row

def get_first_and_last(filename):
    #rdd variable holds the content of file(it's distributed)
    rdd = spark.read.csv(filename, header=True, mode="DROPMALFORMED").rdd

    #Here filename holds abs path. Feel free to substring as per your needs 
    return Row(filename, rdd.first, rdd.take(rdd.count()).last())


spark = SparkSession \
    .builder \
    .appName("Read CSVs") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

# This file list is not distributed one, It holds list of filenames only
filesList = spark.sparkContext\
    .wholeTextFiles("/some-hdfs-path/level-*_var-*/*.csv")\
    .map(lambda x: x[0])\  
    .collect()

#output array
records = filesList.map(get_first_and_last)

for record in records:
    print(record)

我已经在 scala 中尝试了等效代码，并且能够根据需要查看结果。

编辑:根据评论添加了另一种方法。

注意:当使用 sparkContext.wholeTextFiles() 时，小文件是首选，因为每个文件都将完全加载到内存中。 documentation

records = spark.sparkContext\
    .wholeTextFiles("/some-hdfs-path/level-*_var-*/*.csv")\
    .map(lambda x : Row(x[0], x[1].split("\\n")[0], x[1].split("\\n")[-1]))\

for record in records.collect():
    print(record)

pySpark - SparkSession