val aggStream = dataStream
.filter(_.method == "GET")
.keyBy(_.url)
.timeWindow(Time.minutes(10), Time.seconds(5))
.allowedLateness(Time.minutes(1)) //允许延迟
.sideOutputLateData(new OutputTag[ApacheLogEvent]("late"))
.aggregate(new PageCountAgg(), new PageViewCountWindowResult())
当10:14:51的数据来之前,流中最大的事件时间 >= 10:24:50(窗口结束时间) + watermark延迟时间 + 允许延迟时间 ,导致当前waterMark 的值更新为 这个最大的事件时间;导致 10:14:50-10:24:50 的这个窗口关闭。当10:14:51的数据来的时候,因为窗口关闭了,不能放到窗口进行计算了,只能放到侧输出流进行输出了!