第一篇地址 前面一篇只写了思路来源,不是很明确做法. 有些朋友有缘看到了不是很明白,后面换了种简单高效的方式,匆忙记录一下 下面只描述了滚动窗口的做法,滑动窗口做法类似

自定义WindowAssigner

AsyncParamRequestFunction 继承 RichAsyncFunction 使用flink的异步IO获取到窗口大小等动态数据放到对应的元素中. 其中使用了定时过期的Guava Cache来减少数据的请求.具体代码就不贴了, 但是这样也就存在了这些数据一定时间的延迟更新,自己取舍

下面是DynamicTumblingWindowAssigner的代码 完全是复制官方TumblingEventTimeWindows的代码做了一些修改,修改的地方用红框标记出来了. 官方的TumblingEventTimeWindows是通过一个静态方法of(Time size)或者构造函数TumblingEventTimeWindows(long size, long offset)来确定窗口大小.我改成了从assignWindows传入的元素中获取,所以需要在调用.window(DynamicTumblingWindowAssigner)之前就把窗口大小传入元素中. 这个是官方TumblingEventTimeWindows的代码,看不清的直接看源码就好了, 里面有个getDefaultTrigger方法返回值第一个泛型也是Object,所以这个也复制官方的改了下 这个是官方的TumblingEventTimeWindows的getDefaultTrigger中调用的,可以看出就是创建了个 EventTimeTrigger我就把它的EventTimeTrigger完全复制改名为ShockEventEventTimeTrigger, 泛型改成我的ShockEvent就可以了