火花窗口聚合vs. Group By/Join性能

Question

与group by/join相比，我对在窗口上运行聚合函数的性能特征感兴趣。在本例中，我对具有自定义帧边界或顺序的窗口函数不感兴趣，而只是作为运行聚合函数的一种方式。

请注意，我只对大小适中的数据量的批处理（非流式）性能感兴趣，因此我禁用了以下广播连接。

例如，假设我们从以下DataFrame开始：

val df = Seq(("bob", 10), ("sally", 32), ("mike", 9), ("bob", 18)).toDF("name", "age")
df.show(false)

+-----+---+
|name |age|
+-----+---+
|bob  |10 |
|sally|32 |
|mike |9  |
|bob  |18 |
+-----+---+

假设我们想要计算每个名称出现的次数，然后为具有匹配名称的行提供该计数。

val joinResult = df.join(
    df.groupBy($"name").count,
    Seq("name"),
    "inner"
)
joinResult.show(false)

+-----+---+-----+
|name |age|count|
+-----+---+-----+
|sally|32 |1    |
|mike |9  |1    |
|bob  |18 |2    |
|bob  |10 |2    |
+-----+---+-----+

joinResult.explain
== Physical Plan ==
*(4) Project [name#5, age#6, count#12L]
+- *(4) SortMergeJoin [name#5], [name#15], Inner
   :- *(1) Sort [name#5 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(name#5, 200)
   :     +- LocalTableScan [name#5, age#6]
   +- *(3) Sort [name#15 ASC NULLS FIRST], false, 0
      +- *(3) HashAggregate(keys=[name#15], functions=[count(1)])
         +- Exchange hashpartitioning(name#15, 200)
            +- *(2) HashAggregate(keys=[name#15], functions=[partial_count(1)])
               +- LocalTableScan [name#15]

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{functions => f}

val windowResult = df.withColumn("count", f.count($"*").over(Window.partitionBy($"name")))
windowResult.show(false)

+-----+---+-----+
|name |age|count|
+-----+---+-----+
|sally|32 |1    |
|mike |9  |1    |
|bob  |10 |2    |
|bob  |18 |2    |
+-----+---+-----+

windowResult.explain
== Physical Plan ==
Window [count(1) windowspecdefinition(name#5, specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS count#34L], [name#5]
+- *(1) Sort [name#5 ASC NULLS FIRST], false, 0
   +- Exchange hashpartitioning(name#5, 200)
      +- LocalTableScan [name#5, age#6]

根据执行计划，窗口化看起来更有效（阶段更少）。所以我的问题是，情况是否总是如此——我应该一直使用窗口函数来进行这种聚合吗？随着数据的增长，这两种方法的规模会相似吗？那么极端偏斜（即有些名字比其他名字更常见）又如何呢？

Answer 1

禁用广播，因为你的状态和生成一些数据与定时方法1M

生成的计划具有通过窗口进行排序和计数的智能

我尝试了随机的名字样本（省略年龄），得到了这个：

>

加入85.814秒，窗口为50.566秒，窗口为200万记录。集群重启后的计数

加入96.295秒与43.875秒的Windows for 2M记录的竞争。计数

使用的代码：

import scala.collection.mutable.ListBuffer
import scala.util.Random
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{functions => f}

val alpha = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
val size = alpha.size
def randStr(n:Int) = (1 to n).map(_ => alpha(Random.nextInt(size))).mkString

def timeIt[T](op: => T): Float = {
  val start = System.currentTimeMillis
  val res = op
  val end = System.currentTimeMillis
  (end - start) / 1000f
}

var names = new ListBuffer[String]()
for (i <- 1 to 2000000 ) {
    names += randStr(10)     
}
val namesList = names.toSeq
val df = namesList.toDF("name")

val joinResult = df.join(df.groupBy($"name").count, Seq("name"), "inner")
val windowResult = df.withColumn("count", f.count($"*").over(Window.partitionBy($"name")))
val time1 = timeIt(joinResult.count)
val time2 = timeIt(windowResult.count)

println(s"join in $time1 seconds vs $time2 seconds for window")

此外，这个问题表明火花优化器仍然不成熟。

Answer 2

这取决于数据。这里更具体地说，它取决于name列的基数。如果基数较小，则聚合后的数据将较小，并且可以在连接中广播聚合的结果。在这种情况下，连接将比窗口更快。另一方面，如果聚合后基数大，数据大，那么连接将用SortMergeJoin规划，使用窗口将更加高效。

在窗口的情况下，我们有1个总洗牌一个排序。在SortMergeJoin的情况下，我们在左分支中有相同的（总洗牌排序），在右分支中加上额外的减少的洗牌和排序（减少的意思是数据首先聚合）。在连接的右侧分支中，我们还对数据进行了额外的扫描。

另外，你可以查看我在Spark Summit上的视频，我在那里分析了类似的例子。