我想使用Session
窗口聚合,然后Tumble
在生成的结果之上运行窗口聚合Table API/Flink SQL
。
是否可以在第一次聚合后修改rowtime
属性session
以使其等于.rowtime
会话中最后观察到的事件?
我正在尝试做这样的事情:
table
.window(Session withGap 2.minutes on 'rowtime as 'w)
.groupBy('w, 'userId)
.select(
'userId,
('w.end.cast(Types.LONG) - 'w.start.cast(Types.LONG)) as 'sessionDuration,
('w.rowtime - 2.minutes) as 'rowtime
)
.window(Tumble over 5.minutes on 'rowtime as 'w)
.groupBy('w)
.select(
'w.start,
'w.end,
'sessionDuration.avg as 'avgSession,
'sessionDuration.count as 'numberOfSession
)
关键部分是:
('w.rowtime - 2.minutes) as 'rowtime
所以我想重新分配.rowtime
会话中最新事件的记录,没有会话间隙(2.minutes
在本例中)。
这在 BatchTable 中工作正常,但在 StreamTable 中不起作用:
Exception in thread "main" org.apache.flink.table.api.ValidationException: TumblingGroupWindow('w, 'rowtime, 300000.millis) is invalid: Tumbling window expects a time attribute for grouping in a stream environment.
是的,我知道,感觉就像我不想发明时间机器并改变时间的顺序。但实际上有可能以某种方式实现所描述的行为吗?