Spark 开窗函数中order By的作用

大家好，欢迎来到IT知识分享网。

最近几天，受到打击了，总是被人问到 开窗函数中加入 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 和 它的边界值 有关。