sorting - 如何在 Spark DataFrame 上应用部分排序？

Question

以下代码:

val myDF = Seq(83, 90, 40, 94, 12, 70, 56, 70, 28, 91).toDF("number")
myDF.orderBy("number").limit(3).show

输出:

+------+
|number|
+------+
|    12|
|    28|
|    40|
+------+

Spark 的惰性结合 limit 调用和 orderBy 的实现是否会自动导致 partially sorted DataFrame，或者剩余的 7 个数字是否也已排序，即使它不需要？如果是这样，是否有办法避免这种不必要的计算工作？

---

使用 .explain() 显示，执行了两个排序阶段，首先在每个分区上，然后(每个分区前 3 个)一个全局的。但它没有说明这些排序是完整的还是部分的。

myDF.orderBy("number").limit(3).explain(true)

== Parsed Logical Plan ==
GlobalLimit 3
+- LocalLimit 3
   +- Sort [number#3416 ASC NULLS FIRST], true
      +- Project [value#3414 AS number#3416]
         +- LocalRelation [value#3414]

== Analyzed Logical Plan ==
number: int
GlobalLimit 3
+- LocalLimit 3
   +- Sort [number#3416 ASC NULLS FIRST], true
      +- Project [value#3414 AS number#3416]
         +- LocalRelation [value#3414]

== Optimized Logical Plan ==
GlobalLimit 3
+- LocalLimit 3
   +- Sort [number#3416 ASC NULLS FIRST], true
      +- LocalRelation [number#3416]

== Physical Plan ==
TakeOrderedAndProject(limit=3, orderBy=[number#3416 ASC NULLS FIRST], output=[number#3416])
+- LocalTableScan [number#3416]

Answer 1

如果您explain() 您的数据框，您会发现 Spark 将首先在每个分区内执行“本地”排序，然后仅从每个分区中选择前三个元素以进行最终的全局排序前三名。

scala> myDF.orderBy("number").limit(3).explain(true)
== Parsed Logical Plan ==
GlobalLimit 3
+- LocalLimit 3
   +- Sort [number#3 ASC NULLS FIRST], true
      +- Project [value#1 AS number#3]
         +- LocalRelation [value#1]

== Analyzed Logical Plan ==
number: int
GlobalLimit 3
+- LocalLimit 3
   +- Sort [number#3 ASC NULLS FIRST], true
      +- Project [value#1 AS number#3]
         +- LocalRelation [value#1]

== Optimized Logical Plan ==
GlobalLimit 3
+- LocalLimit 3
   +- Sort [number#3 ASC NULLS FIRST], true
      +- LocalRelation [number#3]

== Physical Plan ==
TakeOrderedAndProject(limit=3, orderBy=[number#3 ASC NULLS FIRST], output=[number#3])
+- LocalTableScan [number#3]

我认为它在优化的逻辑计划部分中表现得最好，但物理上也是一样。