Spark-写入128 MB大小的镶木地板文件

Question

我有一个超过10亿行的DataFrame（df）

df.coalesce(5)
.write
.partitionBy("Country", "Date")
.mode("append")
.parquet(datalake_output_path)

从上面的命令中，我了解到我的100个工作节点集群（spark 2.4.5）中只有5个工作节点将执行所有任务。使用聚结剂（5）需要7小时才能完成。

我应该尝试< code >重新分区而不是< code >联合？

有没有一种更快速/高效的方法来写出128 MB大小的拼花文件，或者我需要首先计算数据帧的大小来确定需要多少分区。

例如，如果我的数据帧大小为1 GB，spark.sql.files。maxPartitionBytes=128MB是否应该首先计算否。需要的分区数为1 GB/128 MB=大约（8），然后进行重新分区（8）或合并（8）？

这个想法是在编写时最大限度地扩大输出中镶木地板文件的大小，并且能够快速（更快）做到这一点。

Answer 1

如果您从较高的分区数到较低的分区数，则合并会更好。但是，如果在编写df之前，您的代码没有进行混洗，那么合并将被推到DAG中可能的最早点。您可以做的是在100个分区或任何您认为合适的数字中处理您的df，然后在编写df之前将其持久化。然后使用coalesce将分区减少到5个并写入。这可能会给你更好的表现

Answer 2

您可以获取数据帧< code>df的大小(< code>dfSizeDiskMB)，方法是保存它，然后检查Web UI上的存储选项卡，如本答案所示。有了这些信息和预期的Parquet压缩率的估计值，您就可以估计要达到您想要的输出文件分区大小所需的分区数量

val targetOutputPartitionSizeMB = 128
val parquetCompressionRation = 0.1
val numOutputPartitions = dfSizeDiskMB * parquetCompressionRatio / targetOutputPartitionSizeMB
df.coalesce(numOutputPartitions).write.parquet(path)

请注意，＜code＞spark.files。maxPartitionBytes在这里不相关，因为它是：

读取文件时打包到单个分区中的最大字节数。

（除非 df 是在未创建中间数据帧的情况下读取输入数据源的直接结果。df 的分区数更有可能由 spark.sql.shuffle.partitions 决定，即 Spark 用于从联接和聚合创建的数据帧的分区数）。

我应该尝试重新分区而不是合并吗？

< code>coalesce通常更好，因为它可以避免与< code>repartition相关的混乱，但请注意文档中的警告，即根据您的使用情况，上游阶段可能会失去并行性。