csv - 使用 Spark 将 CSV 转换为 parquet，保留分区

Question

我正在尝试使用 Spark 将一堆 csv 文件转换为 parquet，有趣的是输入 csv 文件已经按目录“分区”。
所有输入文件都具有相同的列集。
输入文件结构如下:

/path/dir1/file1.csv
/path/dir1/file2.csv
/path/dir2/file3.csv
/path/dir3/file4.csv
/path/dir3/file5.csv
/path/dir3/file6.csv

我想用 Spark 读取这些文件并将它们的数据写入 hdfs 中的 Parquet 表，保留分区(按输入目录分区)，例如每个分区有一个输出文件。
输出文件结构应如下所示:

hdfs://path/dir=dir1/part-r-xxx.gz.parquet
hdfs://path/dir=dir2/part-r-yyy.gz.parquet
hdfs://path/dir=dir3/part-r-zzz.gz.parquet

到目前为止，我发现的最佳解决方案是在输入目录之间循环，将 csv 文件加载到数据框中，并将数据框写入 Parquet 表的目标分区中。
但这效率不高，因为我希望每个分区有一个输出文件，写入 hdfs 是一个阻塞循环的单个任务。
我想知道如何以最大的并行性(并且不混洗集群中的数据)来实现这一点。

谢谢 !

Answer 1

重命名您的输入目录更改 dirX至 dir=dirX .然后执行:

spark.read.csv('/path/').coalesce(1).write.partitionBy('dir').parquet('output')

如果您无法重命名目录，则可以使用 Hive Metastore。为每个目录创建外部表和一个分区。然后加载此表并使用上述模式重写。