排序后的数据帧分区数？

Question

spark如何在使用< code>orderBy后确定分区的数量？我一直以为生成的数据帧有< code > spark . SQL . shuffle . partitions ，但这似乎不是真的:

val df = (1 to 10000).map(i => ("a",i)).toDF("n","i").repartition(10).cache

df.orderBy($"i").rdd.getNumPartitions // = 200 (=spark.sql.shuffle.partitions)
df.orderBy($"n").rdd.getNumPartitions // = 2 

在这两种情况下，spark都< code >-Exchange range partitioning(I/n ASC NULLS FIRST，200)，那么第二种情况下的分区数怎么会是2呢？

Answer 1

我进行了各种测试，以便从经验上更仔细地看待这个问题，除了查看排序的范围分区 - 这是这里问题的症结所在。请参阅范围分区程序如何在 Spark 中工作？。

在尝试了“n”的 1 个不同值（如问题中的示例）和 1 个以上的“n”不同值之后，然后使用 df.orderBy（$“n”） 的各种数据帧大小：

• 很明显，用于确定分区数的计算将包含随后通过map分区进行排序的数据范围，
• 它基于在计算这些计算范围的一些启发式最佳分区数之前从现有分区中采样，
• 在大多数情况下，将计算并生成 N 1 个分区，从而将 N 1 分区为空。

分配的额外分区几乎总是空的这一事实使我认为编码中存在某种计算错误，换句话说，恕我直言，这是一个小错误。

我基于以下简单测试，该测试确实返回了我怀疑会认为是正确的分区数的RR：

val df_a1 = (1 to 1).map(i => ("a",i)).toDF("n","i").cache
val df_a2 = (1 to 1).map(i => ("b",i)).toDF("n","i").cache
val df_a3 = (1 to 1).map(i => ("c",i)).toDF("n","i").cache
val df_b = df_a1.union(df_a2)
val df_c = df_b.union(df_a3)

df_c.orderBy($"n")
 .rdd
 .mapPartitionsWithIndex{case (i,rows) => Iterator((i,rows.size))}
 .toDF("partition_number","number_of_records")
 .show(100,false)

退货:

+----------------+-----------------+
|partition_number|number_of_records|
+----------------+-----------------+
|0               |1                |
|1               |1                |
|2               |1                |
+----------------+-----------------+

这个边界示例计算相当简单。只要我对任何一个“N”使用1到2或1…N，就会产生额外的空分区：

+----------------+-----------------+
|partition_number|number_of_records|
+----------------+-----------------+
|0               |2                |
|1               |1                |
|2               |1                |
|3               |0                |
+----------------+-----------------+

排序要求给定“n”或一组“n”的所有数据位于同一分区中。

Answer 2

< code > spark . SQL . shuffle . partitions 用作上限。分区的最终数量是< code>1

正如您所提到的，Spark中的排序经过＜code＞RangePartitioner＜/code＞。它试图实现的是将数据集划分为大致相等范围的指定数量（spark.sql.shuffle.partition）。

可以保证分区后相同的值会在同一个分区中。值得检查Range分区（不是公共API的一部分）类留档：

...

< code>ordering中的表达式计算出相同值的所有行将位于同一分区中

如果不同排序值的数量小于所需的分区数量，即可能范围的数量小于＜code＞spark.sql.shuffle。分区，您将得到较少的分区数。此外，这里引用了＜code＞RangePartitioner＜/code＞Scaladoc的一句话：

在采样记录数小于分区值的情况下，范围分区程序创建的实际分区数可能与 partitions 参数不同。

回到您的示例，n 是一个常量 （“a”），无法进行分区。另一方面，我可以有10，000个可能的值，并被分区为200（=spark.sql.shuffle.partition）范围或分区。

请注意，这仅适用于DataFrame/Dataset API。使用RDD的sortByKey时，可以显式指定分区数，也可以使用Spark当前的分区数。

另请参阅：

• Spark如何实现排序顺序？