如何在spark中使用repartition()指定文件大小

Question

我正在使用 pyspark，并且我有一个大型数据源，我想对其进行重新分区，并明确指定每个分区的文件大小。

我知道使用

repartition(500)

功能会将我的镶木地板分成 500 个大小几乎相等的文件。 问题是每天都会有新数据添加到该数据源中。有些日子可能会有较大的输入，有些日子可能会有较小的输入。因此，当查看一段时间内的分区文件大小分布时，每个文件在

200KB

到

700KB

之间变化。

我正在考虑指定每个分区的“最大大小”，以便无论文件数量多少，每天每个文件的文件大小或多或少相同。 这将帮助我稍后在这个大型数据集上运行我的工作，以避免执行程序时间和洗牌时间等的偏差。 有没有办法使用

repartition()

函数或在将数据帧写入镶木地板时指定它？

Answer 1

maxRecordsPerFile

写入结果。

storage_location = //...
estimated_records_with_desired_size = 2000
result_df.write.option(
     "maxRecordsPerFile", 
     estimated_records_with_desired_size) \
     .parquet(storage_location, compression="snappy")

并且为了确定使用什么作为

maxRecordsPerFile

，您可以获取分区中的所有数据（例如，如果按天分区，则为一天的所有数据），然后除以记录数为了那一天。这是一个简单的

COUNT(*)

与

GROUP BY date

或

WHERE date = 'YYYY-MM-DD'

。假设某一天您总共有 100,000 条记录，总文件大小为 100MB（即 100,000KB），平均每条记录 1KB。