各位读者老爷请放下手上的板砖，我可真没有标题党，且容老弟慢慢道来。
spark和flink本身相信我不用做过多的介绍，后端同学不管搞没搞过大数据，应该都多多少少听过。
如果没听过，简单说，spark和flink之于大数据，就好比vue和react之于前端，就好比spring家族之于java。
从2015年开始接触大数据，2016年开始使用spark，到2022年初能够为spark社区贡献一点微博的贡献，成为spark项目的contributor，对我来说是一段奇特的经历。
这段经历来源于一次spark窗口计算，由于我觉得并不能完全满足要求，想通过源码改造一下。
这一下子进了源码就误入歧途了。
什么要求之类的按不表，进入正题。
什么是窗口计算（可跳过）
在大数据领域，基于窗口的计算是非常常见的场景，特别在于流式计算（flink只不叫实时和离线，只区分有界*）。
如果这说起来还是有点抽象，那举个例子，相信你很快就能明白。
比如微博热搜，我需要每分钟计算过去半小时的热搜词取top50;
比如新能源车，行驶过程中每秒或每两秒钟上报各信号项，如果30秒钟内没有收到该车的信号项，我们认为该车出现故障，便进行预警（假设场景）;
等等。
前者微博热搜是一个典型的时间窗口，后者新能源车是一个典型的会话窗口。
时间窗口(timewindow)
时间窗口又分为滑动窗口（sliding window）和滚动窗口(tumbling window)。反正意思就是这么个意思，在不同的大数据引擎里叫法略有不同，在同一个引擎里不同的API里叫法也略有区别（比如，flink 滑动窗口在DataSet&DataStream api和Table(sql) api里分别叫作sliding window 和HOP window）。
总之时间窗口有一个长度距离(m)和滑动距离(n),当m=n时，这就是一个滚动窗口，相邻窗口两两并不相交重叠。
当m>n时，称为滑动窗口，这时相邻的两个窗口就有了重叠部份。
在多数场景下，m为n的正整数倍。即m%n=0;除非产品经理认为我们应该每61秒统计过去7.3分钟的微博热搜（？？？）。
这个例子可能极端了些，但m%n != 0的实际应用场景肯定是有的。
会话窗口(sessionwindow)
session窗口相对抽象一点。大家可以把session对应到web应用上，理解为一个连接session。
当大数据引擎接收到一条数据相当于一个连接session，当在设定的时间范围内连续没有接收到数据，相当于session会话已断开，这里触发窗口结束。
因此会话窗口长度是不固定的，没有固定的开始和结束。而且相邻的窗口也不会相交重叠。
到这里，大家对大数据的窗口计算应该有了一个简单的感性认识，我们今天讨论的重点是时间窗口，而且只是时间窗口下的一个小小的切点。
今天的主题
即大数据引擎是怎样划分窗口的，当接收到一条数据的时候，数据的时间戳会落到哪些窗口？
先来简单看一点源码。不多，就一点点。
spark获取窗口的主要代码逻辑：
一时看不懂没关系，我第一次看到spark这段代码的时候也有点懵。借助spark的注释来梳理一下。
为了不水字数把spark源代码注释折叠
* The windows are calculated as below:
* maxNumOverlapping <- ceil(windowDuration / slideDuration)
* for (i <- 0 until maxNumOverlapping)
* windowId <- ceil((timestamp - startTime) / slideDuration)
* windowStart <- windowId * slideDuration + (i - maxNumOverlapping) * slideDuration + startTime
* windowEnd <- windowStart + windowDuration
* return windowStart, windowEnd
然后，我们再假设一个简单的场景，将原伪代码进行微调，并配合注释讲解一下。
假设窗口长度（windowDuration ）为10，滑动距离（slideDuration）为5，即每5分钟计算过去10分钟的数据。简单化流程，窗口偏移时间为0。
现在spark集群收到一条数据，它的事件时间戳为13，然后需要计算13会落到哪些窗口里面。
// `获取窗口个数，窗口长度（m）/滑动长度(n)，当两者相等时，就1个窗口; // 当m%n=0时，窗口长度为除数;当m%n!=0时，窗口长度为除数向下的最小整数 // 这里为2个窗口 maxNumOverlapping <- ceil(windowDuration / slideDuration) // 循环获取当前时间戳在每个窗口的边界，即开始时间和结束时间 for (i <- 0 until maxNumOverlapping) // 13/5 -> 2.6 通过ceil向下取整得到2,再+1 = 3 windowId <- ceil(timestamp / slideDuration) // 第1次循环时，计算第1个窗口开始时间 ：3 * 5 +（0 - 2）* 5 = 5 // 第2次循环时，计算第2个窗口开始时间： 3 * 5 + (1 - 2) * 5 = 10 windowStart <- windowId * slideDuration + (i - maxNumOverlapping) * slideDuration // 第1次循环时，计算第1个窗口结束时间：5+10 = 15 // 第2次循环时，计算第2个窗口结束时间：10+10 = 20 windowEnd <- windowStart + windowDuration return windowStart, windowEnd
通过上面的代码，我们知道，时间戳13最终会落到[5-15],[10-20]两个窗口区间。
我们再来看看flink的实现逻辑。
可以看到其实原理类似，先求得窗口个数，略有区别的是，spark是先求得窗口编号windowId，再根据窗口编号求得每一个窗口的开始结束时间。
而spark是直接得到一个窗口开始时间lastWindowStart ，然后根据窗口开始时间+滑动距离=窗口结束时间。
再然后，窗口开始时间-窗口长度=另一个窗口的开始时间，再求得窗口的结束时间。
而不管是哪种方法，都有一个线头。
spark是windowId
windowId <- ceil((timestamp - startTime) / slideDuration)
flink是lastWindowStart .
timestamp - (timestamp - offset + windowSize) % windowSize;
大家发现上面两边代码对比有问题了吗？
spark的两个问题
===========================================5秒钟思考线
点击查看问题答案
问题1：重复计算。`windowId`只需要计算一次就够了。
乃至于`windowStart`也只需要计算一次，根据它，可以计算出当次windowEnd，同样也可以计算出其它的窗口边界。
问题2：ceil和mod(%模运算)的差异。
这两个问题都不是BUG,是性能问题。
第1个问题，直接观察代码就可以得出结论。
第2个问题，需要通过代码测试一下。
因为scala本身也是JVM生态语言，底层都一样。所以我直接使用java写了一个基准测试，内容为ceil和求模的性能差异。
@BenchmarkMode(Mode.AverageTime)
@Warmup(iterations = 3, time = 1)
@Measurement(iterations = 3, time = 4)
@Threads(1)
@Fork(1)
@State(value = Scope.Benchmark)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
public class MathTest {
public static void main(String[] args) throws RunnerException {
Options opt = new OptionsBuilder()
.include(MathTest.class.getSimpleName())
.mode(Mode.All)
.result("MathTest.json")
.resultFormat(ResultFormatType.JSON).build();
new Runner(opt).run();
}
@Benchmark
public void ceil() {
Math.ceil((double)1000/20);
Math.ceil((double)1000/34);
}
@Benchmark
public void mod() {
int a = 1000 % 20;
int b = 1000 % 34;
}
}
大家感兴趣的话，也可以将代码复制到本地，引入JMH就可以运行。
得到的结果：
上图表示的是执行耗时，柱图越高性能越低。更多可以参考我的另一篇文章hashmap的一些性能测试。
那么可以从图中明显看到使用ceil的spark比使用mod的flink在这获取窗口这块功能的性能上肯定要差一些。
不管是第1还是第2个问题，都会随着窗口长度放大。如果我需要每1分钟计算过去60分钟的数据
那么每1条数据进来，它都会进行这样的60次无效计算。
如果一个窗口批次有一万条数据，它就会进行60万次无效计算。
大数据场景下，它的性能损耗是多少呢？
是吧？
既然有性能损耗，它必然可以优化。对于我这种开源爱(投)好(机)者来说，这都是一个大好的机会啊。
原创是不可能原创的，这辈子都不可能，只能靠抄抄flink代码才能成为spark contributor什么的。
好了，我们现在已经学会了两个主流大数据引擎的窗口计算基本原理了，现在我们来写一个大数据引擎吧。
不是，来重（抄）构（袭）spark获取时间窗口的代码吧。
第一个PR
怀着忐忑的心情，给spark社区提了一个PR。想想还有点小激动。
第一次给这种级别的开源社区提交PR。两眼一摸黑，比如PR要怎样写才规范，要不要写test case。要不要@社区大佬等等。
等了两天后，终于有大佬理我了。
我这点渣渣英语，借助翻译软件才完成了PR描述。自然不懂cc是啥，FYI是啥，nit.是啥？看到不认识的自然复制粘贴到翻译软件。
嗯？
现在翻译软件都这样先进了吗？它居然看穿了我是个废物！
第1次提交犯了很多小错误，这是没看源码贡献指南的后果。（其实是看不太懂）
这里面很详细，怎样拉下源码编译，PR标题格式怎样，PR描述规范，代码stylecheck插件等等，事无巨细，是立志于成为spark社区大佬的新手启航必备。
如果英语老师没有骗我， could you please表示的应该是委婉，客气。
社区大佬都很有礼貌，说话又好听，我超喜欢的。
If you don't mind, could you please
在这位大佬发现我没做性能测试后，（真的，现在想想，性能改进的代码没有基准测试你敢信），温柔提醒我，在看穿我是个新（废）手（物）后，亲自写了benchmark基准测试代码。并得出新的计算逻辑比原有的性能提升30%到60%。
然后经过一番沟通修改代码注释什么的，最终合并到了master.
尾声
以上就是我的第一次spark PR之旅了。
如果你要问我感受的话。短暂的兴奋过后就是空虚。
不玩梗地说，spark社区氛围真的很好。在后面陆陆续续又给sparkT和flink提交了几个PR。没有对比就没有伤害，比起flink,spark真的对新手非常友好了。
这过程中，踩了很多坑。也收获很多。比如，这次30%到60%的性能提高，对于一个比较成熟的大数据产品来说，应该算是比较大的提升了吧？
但在后面的PR中，我做出远超此次的性能提升，而且不是借助flink的既有逻辑，完全独立完成。
后面有时间也可以把这些写出来，水水文章。
彩蛋
本来到这里就结束。但是，偶然在flink社区PR区看到一个熟悉timewindow,哟呵，这个我熟啊。
点进去一看，尴尬了。
大家看一下，这次PR提交的主要代码更改逻辑就知道了。
没错，就是之前我给spark 提交的代码的借（抄）鉴（袭）来源。flink时间窗口分配窗口的核心代码。而且这不是优化，而是修复BUG。
哦，原来这特么的是彩蛋，这特么的是惊喜啊！
这就好比，照着隔壁班里第一名抄作业，老师给了个高分，然后被高手自爆，老师，我写得有问题。
它是怎炸的呢？
原文已经说得非常清楚，我在这里长话短说。
简单画了个图：
假设时间戳13在一个长度15，滑动长度5的窗口逻辑里，我们要知道它会分配到哪2个窗口里，只需求得最后一个窗口开始的长度即可。
它最后一个窗口的开始长度为 13%5 = 3,为时间戳对滑动距离求模。即把上图中红色部份减去，或者向左偏移余数部份，就是它最后一个窗口的开始长度。
不管怎样，时间戳必须得落到开始时间后面，窗口必须包含时间戳。
好，很好，很有精神。没有问题!no problem!
但是！如果时间戳是负数呢？比如-1呢？
我们开始求它的最后一个窗口开始时间，时间戳对滑动距离求模，即-1%5 = -1。
-1 - (-1) = 0
这样就导致不管是-1还是13都应该向左偏移的，结果跑向右边了。
13:???
开始时间大于了时间戳本身，时间戳跑到窗口外面去了，这肯定是不正常的。
其实不仅仅是负的时间戳，是(timestamp - window.starttime)% window.slideduration <0的情况下都会有这种问题。
只说负的时间戳有问题，就显得我的上个PR很无脑。显得我无脑没关系，这其实也是小看了spark,flink这种大范围流行的开源框架。
通过spark的测试案例也能很清楚的看到，肯定是考虑到了时间戳落到1970-01-01之前的。
随手截一个测试案例
只不过它的时间戳都在1970-01-01前后几秒钟范围，落在了滑动距离之内。所以这个问题没有及时暴露出来。
而且在我提交优化的PR之前，spark本身的代码是不会出现这种问题的。所以这个锅完全是我的，必须背了。
然后给社区提交了一个fix
由于种种原因，我倒是放了鸽子了。后来，被另一哥们重新提交合并。
https://github.com/apache/spark/pull/39843#issuecomment-1418436041
(完)