Scala - Spark Repartition 没有给出预期的结果

Question

我想根据 X 列重新分区我的 spark 数据框。假设 X 列有 3 个不同的值（X1、X2、X3）。不同值的数量可能会有所不同。

我希望一个分区包含只有 1 个 X 值的记录。 IE。我想要 3 个分区，其中 1 个的记录是 X=X1 ，另一个是 X=X2 ，最后一个是 X=X3。

我通过查询从数据帧中获得了 X 的独特值

val uniqueList = DF.select("X").distinct().map(x => x(0).toString).collect() 

正确地给出唯一值列表。

我正在重新分区

DF = DF.repartition(uniqueList.length, col('X'))

但是，我在 DF 中的分区没有按预期进行。数据分布不正确，一个分区为空，第二个分区包含 X1 的记录，第三个分区包含 X2 和 X3 的记录。

如果我遗漏了什么，有人可以帮忙吗？

编辑：

我的 X 列可能有不同数量的唯一值。它可能有 3 或 3000 个唯一值。 如果我在下面这样做

DF = DF.repartition(col('X'))

我只会得到 200 个分区，因为这是 spark.sql.shuffle.partitions 的默认值。因此我给出了分区数

如果 X 有 3000 个唯一值，那么我想对我的 DF 进行重新分区，使得有 3000 个分区并且每个分区包含 X 的一个特定值的记录。这样我就可以运行 mapPartition 并并行处理每个分区.

Answer 1

重分区是基于hash分区（取partitioning key的hash码取partitioning个数为模），所以每个partition是否只有一个值纯属偶然。

如果您可以将每个分区键映射到零到（唯一值的数量 - 1）范围内的唯一 Int，因为 Scala 中 Int 的哈希码是该整数，这将确保如果分区的数量至少与唯一值的数量一样多，则没有分区具有多个不同的分区键值。

也就是说，为此类Ints 分配值本质上是不可并行的，并且需要顺序扫描或提前知道不同的值。

在概率上，特定值散列到（n 个分区）的特定分区中的机会是 1/n。随着 n 相对于不同值的数量增加，没有分区具有多个不同值的机会增加（在极限情况下，如果您可以拥有 2^32 个分区，则几乎所有分区都是空，但实际的哈希冲突仍会保证分区中有多个不同的值）。因此，如果您可以容忍空分区，则选择足够多于不同值数量的分区数将减少次理想结果的机会。

Answer 2

您的 coln X 是否包含空值？然后 Spark 尝试为此创建一个分区。由于您还将分区数指定为 int，因此 Spark 可能会尝试压缩 X2 和 X3。因此，您可以尝试两件事 - 只需给出用于修复的列名（仍然是一个额外的分区）或尝试从 X 中删除空值（如果存在）。

Answer 3

这行得通吗？

val repartitionedDF = DF.repartition(col("X"))

这是我在博客上写的一个例子

数据：

first_name,last_name,country
Ernesto,Guevara,Argentina
Vladimir,Putin,Russia
Maria,Sharapova,Russia
Bruce,Lee,China
Jack,Ma,China

代码：

df
  .repartition(col("country"))
  .write
  .partitionBy("country")
  .parquet(outputPath)

文件系统输出：

partitioned_lake1/
  country=Argentina/
    part-00044-cf737804-90ea-4c37-94f8-9aa016f6953a.c000.snappy.parquet
  country=China/
    part-00059-cf737804-90ea-4c37-94f8-9aa016f6953a.c000.snappy.parquet
  country=Russia/
    part-00002-cf737804-90ea-4c37-94f8-9aa016f6953a.c000.snappy.parquet