如何在使用spark数据帧写入时自动计算numRepartition

Question

当我尝试将dataframe写入Hive拼接分区表时

df.write.partitionBy("key").mode("append").format("hive").saveAsTable("db.table")

它会在HDFS中创建大量的块，每个块只有很小的数据量。

我理解它是如何进行的，因为每个spark子任务都会创建一个块，然后向其中写入数据。

我也理解，块的数量会提高Hadoop的性能，但达到阈值后也会降低性能。

如果我想自动设置numPartition，有没有人有好主意？

numPartition = ??? // auto calc basing on df size or something
df.repartition("numPartition").write
  .partitionBy("key")
  .format("hive")
  .saveAsTable("db.table")

Answer 1

首先，当您已经在使用partitionBy(key)时，为什么还要有一个额外的重新分区步骤-您的数据将根据键进行分区。

通常，您可以按列值重新分区，这是一个常见的场景，有助于reduceByKey、基于列值的过滤等操作。例如，

val birthYears = List(
  (2000, "name1"),
  (2000, "name2"),
  (2001, "name3"),
  (2000, "name4"),
  (2001, "name5")
)
val df = birthYears.toDF("year", "name")

df.repartition($"year") 

Answer 2

默认情况下，spark将为随机操作创建200个分区。因此，200个文件/块(如果文件大小较小)将写入HDFS。

根据您在Spark中的数据配置shuffle后要创建的分区数量，使用以下配置：

spark.conf.set("spark.sql.shuffle.partitions", <Number of paritions>)

例如：spark.conf.set("spark.sql.shuffle.partitions", "5")，所以Spark将创建5个分区，并将5个文件写入HDFS。