如何使用 pyspark 提高 csv 到 parquet 文件格式的性能？答案

【问题标题】：How to improve performance of csv to parquet file format using pyspark?如何使用 pyspark 提高 csv 到 parquet 文件格式的性能？
【发布时间】：2021-05-02 09:47:36
【问题描述】：

我有一个大型数据集，需要使用 pyspark 从 csv 转换为 parquet 格式。大约有 500GB 的数据分散在数千个 csv 文件中。我最初的实现很简单......

spark = SparkSession.builder \
    .master("local") \
    .appName("test") \
    .getOrCreate()

df = spark.read.csv(input_files, header=True, inferSchema=True)

df.repartition(1).write.mode('overwrite').parquet(output_dir)

性能太差了，我已经让它运行了 2 个多小时才放弃。从日志输出我推断它甚至没有完成将 csv 文件读入数据帧。

我在具有 128 个高性能 CPU 内核和 1TB 内存的服务器上本地运行 spark。磁盘存储基于 SSD，已确认读取速度为 650 MB/s。我的直觉是，考虑到可用的计算资源，我应该能够显着提高性能。我正在寻找有关如何执行此操作的提示。

我试过了……

不推断架构，这并没有产生明显的性能差异（架构是四列文本）
使用配置设置spark.executor.cores 来匹配我服务器上的物理内核数。设置似乎没有任何效果，我没有观察到使用更多内核的系统。

根据管理方向，我暂时无法使用 pyspark，但如有必要，我可以说服他们使用不同的工具。

【问题讨论】：

您确定要一个没有任何分区的 Parquet 文件吗？具有分区列应该会加快速度，不仅在写入方面，而且在未来读取和执行计算方面。
@mck 我已经考虑过，确定最佳分区数的最佳方法是什么？
这是一个非常广泛的问题——谷歌会给你很多答案。这真的取决于你的数据结构。
还提供架构 - 手动或按名称读取一个 CSV 文件，并使用 inferSchema 读取它，然后使用该文件中的架构读取所有文件

标签： performance csv pyspark parquet

【解决方案1】：

根据我使用 spark 的经验提出的一些建议：

如果您正在处理大量数据，则不应推断架构。它可能不会显着提高性能，但肯定会为您节省一些时间。
不要使用 repartition(1)，因为它会打乱数据并使用数据创建单个分区，而对于拥有的大量数据，这是您不希望的。我建议您尽可能根据您的集群配置增加分区数量，以便更快地保存 parquet 文件。
如果您只是读取 csv 文件，然后在下一步将其保存为 parquet 文件，请不要缓存/保留您的数据帧。它可以增加您的节省时间，因为缓存本身需要一些时间。如果您在数据帧上执行多个转换然后对其执行多个操作，那么缓存数据帧会有所帮助。您只执行将数据帧写入镶木地板文件的一项操作，因此根据我的说法，您不应该缓存数据帧。

【讨论】：

【解决方案2】：

一些可能的改进：

不要使用.repartition(1)，因为你会失去写操作的并行性
在写入前保留/缓存数据帧：df.persist()

如果您确实需要将其保存为 1 个 parquet 文件，您可以先写入临时文件夹而不减少分区，然后在第二次写入操作中使用 coalesce：

df = spark.read.csv(input_files, header=True, inferSchema=True).persist()
# ....

df.write.mode('overwrite').parquet("/temp/folder")
df.unpersist()

df1 = spark.read.parquet("/temp/folder")
df1.coalesce(1).write.mode('overwrite').parquet(output_dir)

【讨论】：

关于重新分区的好建议，我的用例没有必要。我在一个例子中看到它并认为它是强制性的。