​

然后我们再来看:在事件时间语义下,我们来测试一下窗口,之前我们都不是用的这个EventTime对吧

之前我们用的是,processTime,处理时间对吧,也不是用的,摄入时间.

现在我们用eventTime事件时间来,进行一下窗口操作,首先我们还是用

WindowTest1_TimeWindow1这个测试类,可以看到上面

我们已经配置好了,环境是用事件时间,然后设置了自动waterMark,的周期是100毫秒.

然后我们从socket流中去读取数据.

​

然后我们再来看,我们拿到原始的 inputStream 流以后,我们这里转换成SensorReading 流

通过map方法

​

然后我们拿到map以后的,dataStream流,以后,我们再去给这个

​

map后的SensorReading流,再去添加waterMark,可以看到我们提取了,我们数据中时间做为时间戳,

这个单位是秒,所以*1000对吧

​

然后我们拿到这个dataStream以后,首先.keyBy(“id”)

我们进行了分组,分组以后,我们去开窗,这里我们设置15秒的窗口时间,

设置以后,然后我们再进行minBy,根据temperature,也就是根据温度值进行minBy

,这里业务上,我们就是获取的,15秒钟以内的最小值对吧.

注意因为我们用了keyBy,所以数据被分组了,所以上面这里,我们获取的是15秒内,

每个分组,中的最小的温度值.

​

可以看到上面的代码

​

得到处理后的流以后,然后我们直接.print(“minTemp”)

打印结果

​

然后我们启动程序,然后再启动netcat去测试,可以看到nc -k 7777

我们监听7777端口

​

然后我们这里,去输入数据看看,首先输入第一条数据

​

我们继续输入数据,可以看到没有出来结果,因为需要等15秒,窗口才关闭,才会输出结果对吧.

但是我们想一想,现在我们已经过了15秒的窗口时间了,好像还是没有输出结果对吧.

怎么回事?

​

我们先看,我们设置waterMark,这里是延迟2秒对吧,而我们数据的单位就是秒,所以

这里1547718205这个时间的时候,实际上这时候的waterMark是1547718203对吧

因为我们的waterMark延迟了2秒.

​

然后再输入1547718206,这个时间数据

​

然后继续输入,1547718212 可以看到,数据,输入到这里以后

接着结果就出来了 而且可以看到出来了4条对吧.

为什么是4条数据,注意,因为,我们对数据进行了keyBy分组对吧.

​

可以看到,这里每个分组这里,sensor_1,sensor_6,sensor_10 sensor_7

这里4个组,每个组,到15秒的时候都会关窗口对吧.

所以就输出了4个数据,每个数据,都输出了该组中的最小的温度值对吧,

sensor_1这个看的比较明显

​

然后我们现在再来看,现在的时间可以看到是1547718212对吧,那么

现在结果出来了,那么对应的窗口是多少呢?

因为waterMark延迟2秒,所以,这个时候的关闭的窗口应该是:

1547718212 – 2 = 1547718210 对吧

所以这个窗口就是 [195,210)

​

然后我们知道现在出结果的窗口是[195,210)了,以后然后

[210,225) 我们再看这个数据,这个是我们推测出来的,下一个窗口的时间段.

就是从210开始到225,不包括225,包括224对吧,这个是下一个窗口.

​

然后我们再来输入数据,可以看到当我们时间戳输入到

1547718227 这里的时候,结果突然就输出了,这时候

waterMark是,1547718227 – 2 =1547718225对吧,

但是这个时候,我们说,窗口是多少 [210,225) 对吧

但是前闭后开,不包含225对吧.

所以是[210,224]

然后我们看到210到225之间的数据,最小值也是32.1也没有错对吧.

​