TIMESTAMP(3)/TIMESTAMP_LTZ(3)#xff0c;且是

sche…WATERMARK

把一个现有的列定义为一个为表标记事件时间的属性。

该列的类型必须为

TIMESTAMP(3)/TIMESTAMP_LTZ(3)且是

schema

watermark就会触发当前行数据的计算watermark的形象描述如下

watermark会根据数据流中event的时间戳发生变化。

通常情况下event都是乱序的不按时间排序的。

watermark的计算逻辑为当前最大的

event

最大允许延迟时间MaxOutOfOrderness。

在同一个分区内部当watermark大于或者等于窗口的结束时间时才能触发该窗口的计算即watermarkwindows

根据上图分析

窗口需要在WM12:15时才能被触发计算参与计算的event为event(12:07)/event(12:01)/event(12:07)/event(12:09)event(12:10)/event(12:12)/event(12:12)/event(12:13)/event(12:20)/event(12:14)/event(12:15)不参与计算因为还未到窗口时间也就是event

time

窗口内的event就算超过了最大允许延迟时间MaxOutOfOrderness不会再参与计算也就是数据被强制丢掉了。

对于

[12:10,12:20]

窗口会在WM12:30时被同时触发计算参与**[12:10,12:20]**

窗口计算的event为event(12:10)/event(12:12)/event(12:12)/event(12:13)/event(12:14)/event(12:15)/event(12:15)/event(12:18)参与

[12:20,12:30]

窗口计算的event为event(12:20)/event(12:20)在这个过程中event(12:05)会被丢弃不会参与计算因为已经超了最大允许延迟时间MaxOutOfOrderness

迟到的事件的处理在介绍watermark时提到了现实中往往处理的是乱序event即当event处于某些原因而延后到达时往往会发生该event

time

watermark的情况所以flink对处理乱序event的watermark有一个允许延迟的机制这个机制就是最大允许延迟时间MaxOutOfOrderness允许在一定时间内迟到的event仍然视为有效event。

WATERMARK

--计算列必须为TIMESTAMP(3)/TIMESTAMP_LTZ(3)类型WATERMARK

FOR

kafka,properties.bootstrap.servers

***,topic

--事件时间属性字段类型为TIMESTAMP(3)WATERMARK

FOR

TIMESTAMP(3)/TIMESTAMP_LTZ(3)WATERMARK

FOR

kafka,properties.bootstrap.servers

***,topic

--RANGE每个group_name计算当前group_name前10分钟内收到的同一group_name的所有总数

selectgroup_name

--求每10分钟的滚动窗口内同一group_name的所有总数

create

window_start,window_end,window_time,group_name,count(*)

cnt

window_start,window_end,window_time,group_name参考

Window