Spark 2.4窗口函数中orderBy对collect_set聚合的行为疑问

问题原因 #

当你在Window中添加orderBy子句但未显式指定**窗口范围（Frame）**时，Spark会默认使用RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW作为窗口范围。这意味着collect_set只会聚合从分区第一行到当前行的元素，从而产生累积式的结果。

而不带orderBy时，Spark默认的窗口范围是RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING，即覆盖整个分区，所以能得到全分区聚合的结果。

另外需要注意：collect_set本身是无序集合，即使结合orderBy，Spark也不保证集合内元素的顺序。如果需要严格控制顺序，建议使用collect_list先收集有序元素，再去重。

解决方法 #

方法1：显式指定全分区窗口范围（保留collect_set） #

通过rowsBetween显式设置窗口范围覆盖整个分区，同时添加orderBy控制聚合时的元素顺序：

val data = Seq(("opp1","y","p1"),("opp1","y","p2"),("opp1","n","p3"),("opp2","y","p1"),("opp2","y","p4"),("opp2","n","p5"),("opp2","n","p6"),("opp2","n","p7"))
val df = data.toDF("opp", "stock", "product")

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

// 定义包含全分区范围和排序的窗口
val fullPartitionWindow = Window.partitionBy("opp", "stock")
  .orderBy(col("product").asc)
  .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)

// 使用collect_set进行全分区聚合
df.withColumn("product_stock_status", collect_set(col("product")).over(fullPartitionWindow))
  .show()

执行结果中，每个分区的所有行都会得到相同的全聚合集合，元素顺序会受orderBy影响（注意：collect_set的顺序不被官方保证，仅在当前场景下表现有序）。

方法2：用collect_list+array_distinct实现严格有序去重 #

如果需要严格保证集合元素的顺序，先通过collect_list收集排序后的元素，再用array_distinct去重（Spark 2.4支持此函数）：

val data = Seq(("opp1","y","p1"),("opp1","y","p2"),("opp1","n","p3"),("opp2","y","p1"),("opp2","y","p4"),("opp2","n","p5"),("opp2","n","p6"),("opp2","n","p7"))
val df = data.toDF("opp", "stock", "product")

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

val fullPartitionWindow = Window.partitionBy("opp", "stock")
  .orderBy(col("product").asc)
  .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)

df.withColumn("sorted_product_list", collect_list(col("product")).over(fullPartitionWindow))
  .withColumn("product_stock_status", array_distinct(col("sorted_product_list")))
  .drop("sorted_product_list")
  .show()

此方法会得到严格按product升序排列的去重集合，且每个分区的所有行结果一致。

内容的提问来源于stack exchange，提问作者SD_