有没有办法更改每个分区文件夹的输出行数？

Question

我有大量最终结果数据，这些数据在我感兴趣的列中分布不均。当我直接通过分区写入时，每个分区的文件数与 spark.sql.shuffle.partitions 相同。这导致拥挤分区中的每个文件都非常大（以 GB 为单位），但在其他一些分区中，文件大小非常小（甚至以 KB 为单位）。有没有办法改变每个分区的文件数？

例子：

+----------------------------+----------+
| number of rows in category | category |
+----------------------------+----------+
| 50000000000                |    A     |
| 200000                     |    B     |
| 30000                      |    C     |
+----------------------------+----------+

如果我这样做：

df.write.partitionBy("category").parquet(output_dir)

文件夹“A”中的文件很大，而“B”和“C”中的文件很小。

Answer 1

尝试使用多列重新分区数据框（如果可能并且对您的数据合乎逻辑）。

例子：

df.repartition("category", "<some_other_column_name>").write.partitionBy("category").parquet(output_dir)

Answer 2

我建议在数据帧上调用 df.repartition(NUM_PARTITIONS) 以在分区上均匀分布行。在您的情况下，对于类别 = A，与类别 C 相比，行将分布在更多数量的分区上。重新分区后，当您为类别 A 调用 write.partitionBy("category") 时，因为它分布在更多分区上，所以更多数量的文件将被写入（每个 A 类分区一个文件）。

NUM_PARTITIONS 可以是动态的，例如 NUM_PARTITIONS = df.count()%ROWS_PER_PARTITION。您可以根据每行的字节大小来决定多少 ROWS_PER_PARTITION。

NUM_PARTITIONS = 100 
df.repartition(NUM_PARTITIONS).write.partitionBy("category").parquet(output_dir)

如果要查看分区是如何分布的，可以使用这个

import pyspark.sql.functions as f
df.withColumn("partition_id",f.spark_partition_id()).groupBy("partition_id").count().show()

更详细的讨论，你看这个Spark SQL - Difference between df.repartition and DataFrameWriter partitionBy?