apache-spark - 加载 Parquet 文件并保留相同数量的hdfs分区

Question

我有一个 Parquet 文件/df保存在具有120个分区的hdfs中。 hdfs上每个分区的大小约为43.5M。

总尺寸

hdfs dfs -du -s -h /df
5.1 G  15.3 G  /df

hdfs dfs -du -h /df
43.6 M  130.7 M  /df/pid=0
43.5 M  130.5 M  /df/pid=1
...
43.6 M  130.9 M  /df/pid=119

我想将该文件加载到Spark中并保持相同数量的分区。
但是，Spark将自动将文件加载到60个分区中。

df = spark.read.parquet('df')
df.rdd.getNumPartitions()

60

HDFS设置:

未设置 'parquet.block.size'。

sc._jsc.hadoopConfiguration().get('parquet.block.size')

什么也不返回。

'dfs.blocksize'设置为128。

float(sc._jsc.hadoopConfiguration().get("dfs.blocksize"))/2**20

退货

128

将这些值中的任何一个更改为较低值都不会导致 Parquet 文件加载到hdfs中相同数量的分区中。

例如:

sc._jsc.hadoopConfiguration().setInt("parquet.block.size", 64*2**20)
sc._jsc.hadoopConfiguration().setInt("dfs.blocksize", 64*2**20)

我意识到43.5 M远低于128M。但是，对于此应用程序，我将立即完成许多转换，这将导致120个分区中的每个分区都更接近128M。

我试图避免自己在加载后不得不重新分区到应用程序中。

有没有一种方法可以强制Spark使用与hdfs中存储的分区数量相同的分区来加载 Parquet 文件？

Answer 1

首先，我将从检查Spark如何将数据拆分为分区开始。
默认情况下，它取决于数据和群集的性质和大小。
本文应该为您提供答案，为什么您的数据帧被加载到60个分区中:

https://umbertogriffo.gitbooks.io/apache-spark-best-practices-and-tuning/content/sparksqlshufflepartitions_draft.html

通常，它的 Catalyst 负责所有优化(包括分区数)，因此，除非确实有充分的理由进行自定义设置，否则我会让它完成其工作。如果您使用的任何转换范围很广，Spark都会反洗数据。