Spark 开窗函数中order By的作用（spark窗口函数原理）

最近几天，受到打击了，总是被人问到 开窗函数中加入 order by 和 不加 order by 有什么区别。

例如 sum(x1) over (partition by x2 order by x1) 和 sum(x1) over (partition by x2)的结果一样吗？为什么？

我一开始觉得，你在 sum中后面加什么order by，反正都是获取累加的结果，加不加order by有什么影响？

后来我自己尝试运行，看了结果，的确不一样，怎么会这样，给大家看看结果：

这是输入数据：
+---+---+
| id| kk|
+---+---+
|  0|  1|
|  1|  1|
|  2|  1|
|  3|  2|
|  4|  2|
|  5|  2|
|  6|  2|
|  7|  3|
|  8|  3|
|  9|  3|
+---+---+
  
这是执行了  functions.sum("kk").over(Window.partitionBy("kk").orderBy("id")
+---+---+---+
| id| kk| dd|
+---+---+---+
|  0|  1|  1|
|  1|  1|  2|
|  2|  1|  3|
|  7|  3|  3|
|  8|  3|  6|
|  9|  3|  9|
|  3|  2|  2|
|  4|  2|  4|
|  5|  2|  6|
|  6|  2|  8|
+---+---+---+

+---+---+---+
| id| kk| dd|
+---+---+---+
|  0|  1|  3|
|  1|  1|  3|
|  2|  1|  3|
|  7|  3|  9|
|  8|  3|  9|
|  9|  3|  9|
|  3|  2|  8|
|  4|  2|  8|
|  5|  2|  8|
|  6|  2|  8|
+---+---+---+

大家可以看到，不一样

为什么呢？后面我是翻了spark的源码 和 看了执行计划才得到的答案。

.withColumn("dd", functions.sum("kk").over(Window.partitionBy("kk").orderBy("id") ) 它的物理执行计划是：

Window [sum(cast(kk#2 as bigint)) windowspecdefinition(kk#2, id#0L ASC NULLS FIRST, specifiedwindowframe(RangeFrame, unboundedpreceding$(), currentrow$())) AS dd#13L], [kk#2], [id#0L ASC NULLS FIRST]

.withColumn("dd", functions.sum("kk").over(Window.partitionBy("kk"))) 它的物理执行计划是：

Window [sum(cast(kk#2 as bigint)) windowspecdefinition(kk#2, specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS dd#18L], [kk#2]

从这里就可以发现猫腻，

加入了order by之后，spark对于开窗函数，windowspecdefinition 选择了 RangeFrame，并且它的边界是 第一行 到当前行。

不加入order by之后，spark对于开窗函数，windowspecdefinition 选择了 RowFrame，并且它的边界是第一行到最后一行。

所以我们看到 加入order by之后的结果就是

+---+---+---+
| id| kk| dd|
+---+---+---+
|  0|  1|  1| 1
|  1|  1|  2| 1+1
|  2|  1|  3| 1+1+1
|  7|  3|  3| 3
|  8|  3|  6| 3+3
|  9|  3|  9| 3+3+3
|  3|  2|  2| 2frame
|  4|  2|  4| 2+2
|  5|  2|  6| 2+2+2
|  6|  2|  8| 2+2+2+2
+---+---+---+

因为结果都是首行到当前行的累加值。

那么结果很明确了，就是 加入 order by之后，spark底层选择 frame type会是 RangeFrame，并且默认它的边界就是 unboundedpreceding , currentrow。这就是加入order by的作用，之所以累加和不同，就是这个 RangeFrame和它的边界值搞得鬼。

那反过来，如果我还是使用 order by，但是我手动设置它的边界值，是 unboundedpreceding 和

unboundedfollowing，那样，执行的结果应该和 不加入order by的结果应该一样。

所以我又尝试了

.withColumn("dd", functions.sum("kk").over(Window.partitionBy("kk").orderBy("id")
    .rangeBetween(Long.MinValue,Long.MaxValue)
)
自己手动设定它的边界值，
运行结果就是
+---+---+---+
| id| kk| dd|
+---+---+---+
|  0|  1|  3|
|  1|  1|  3|
|  2|  1|  3|
|  7|  3|  9|
|  8|  3|  9|
|  9|  3|  9|
|  3|  2|  8|
|  4|  2|  8|
|  5|  2|  8|
|  6|  2|  8|
+---+---+---+

真的，这就是 真相了，加入 order by 和 不加入 order by的真相。

多看源码，其实和 底层 选择的 Frame 和 它的边界值 有关。