做哪种网站赚钱,动漫网站怎么建设,南昌网站搭建服务,广州白云发布最新通告分析回答
Flink的窗口机制是其底层核心之一#xff0c;也是高效流处理的关键。Flink窗口分配的基类是WindowAssigner抽象类#xff0c;下面的类图示出了Flink能够提供的所有窗口类型。 Flink窗口分为滚动#xff08;tumbling#xff09;、滑动#xff08;sliding回答
Flink的窗口机制是其底层核心之一也是高效流处理的关键。Flink窗口分配的基类是WindowAssigner抽象类下面的类图示出了Flink能够提供的所有窗口类型。 Flink窗口分为滚动tumbling、滑动sliding和会话session窗口三大类本文要说的是滑动窗口。
下图示出一个典型的统计用户访问的滑动窗口来自官方文档。 假设每两条虚线之间代表1分钟时间差那么窗口大小size就是2分钟滑动步长slide是1分钟。若时间特征为事件时间代码如下。
dataStream .keyBy(userId) .window(SlidingEventTimeWindows.of(Time.minutes(2), Time.minutes(1))); 由图可知当前滑动窗口与上一个滑动窗口会有重叠。在窗口大小size是步长slide的2倍的情况下几乎每个DataStream元素都会处于2个窗口内。
我们简单参考一下相关的Flink源码以加深理解。以下是窗口算子WindowOperator的processElement()方法的部分源码。 Overridepublic void processElement(StreamRecordIN element) throws Exception {final CollectionW elementWindows windowAssigner.assignWindows(element.getValue(), element.getTimestamp(), windowAssignerContext);boolean isSkippedElement true;final K key this.KgetKeyedStateBackend().getCurrentKey();if (windowAssigner instanceof MergingWindowAssigner) {// 会话窗口的处理逻辑略去} else {for (W window : elementWindows) {if (isWindowLate(window)) {continue;}isSkippedElement false;windowState.setCurrentNamespace(window);windowState.add(element.getValue());triggerContext.key key;triggerContext.window window;TriggerResult triggerResult triggerContext.onElement(element);if (triggerResult.isFire()) {ACC contents windowState.get();if (contents null) {continue;}emitWindowContents(window, contents);}if (triggerResult.isPurge()) {windowState.clear();}registerCleanupTimer(window);}}// 最后是侧输出迟到数据的逻辑略去}
复制代码
该方法先调用WindowAssigner.assignWindows()方法根据输入元素的时间戳判断它应该属于哪些窗口。接着遍历所有窗口将该元素加入对应的窗口状态即缓存中并根据触发器返回的TriggerResult决定是输出fire还是清除purge窗口的内容emitWindowContents()方法会调用用户函数。最后还要调用registerCleanupTimer()方法注册计时器用来在窗口彻底过期时清除窗口状态。
以下是SlidingEventTimeWindows.assignWindows()方法的源码。 Overridepublic CollectionTimeWindow assignWindows(Object element, long timestamp, WindowAssignerContext context) {if (timestamp Long.MIN_VALUE) {ListTimeWindow windows new ArrayList((int) (size / slide));long lastStart TimeWindow.getWindowStartWithOffset(timestamp, offset, slide);for (long start lastStart;start timestamp - size;start - slide) {windows.add(new TimeWindow(start, start size));}return windows;} else {throw new RuntimeException(Record has Long.MIN_VALUE timestamp ( no timestamp marker). Is the time characteristic set to ProcessingTime, or did you forget to call DataStream.assignTimestampsAndWatermarks(...)?);}}public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {return timestamp - (timestamp - offset windowSize) % windowSize;}
复制代码
这段代码就不难理解了先调用getWindowStartWithOffset()方法根据元素的时间戳计算出其窗口的起点时间戳再逐次循环向后滑动产生size / slide个窗口。我们可以将size / slide叫做“粒度”亦即上述代码中返回的Collection集合的大小。粒度越大“细”滑动窗口之间的重合也越大。
代码读完了有一个貌似稀松平常的需求
以3分钟的频率实时计算App内各个子模块近24小时的PV和UV。
直觉上我们需要用粒度为1440 / 3 480的滑动窗口来实现它但是细粒度的滑动窗口会带来性能问题有两点
状态 由代码可知WindowOperator内维护了窗口本身的内部状态windowState类型为InternalAppendingState。对于一个元素会将其写入对应的(key, window)二元组所圈定的状态中。可见如果粒度为480那么每个元素到来更新windowState时都要遍历480个窗口并写入开销是非常大的。在采用HDFS/RocksDB作为状态后端时checkpoint的瓶颈也尤其明显。
定时器 在Flink中定时器的实际实现是TimerHeapInternalTimer类并且是用Flink自己实现的优先队列维护在堆内存中的。而在WindowOperator中每一个(key, window)二元组都需要注册两个定时器一是触发器注册的定时器用于决定窗口数据何时输出二是registerCleanupTimer()方法注册的清理定时器用于在窗口彻底过期如allowedLateness过期之后及时清理掉窗口的内部状态。细粒度滑动窗口会造成维护的定时器增多内存负担加重。
在官方文档Windows最后一节的最后也有如下的提醒
Flink creates one copy of each element per window to which it belongs. Given this, tumbling windows keep one copy of each element (an element belongs to exactly one window unless it is dropped late). In contrast, sliding windows create several of each element, as explained in the Window Assigners section. Hence, a sliding window of size 1 day and slide 1 second might not be a good idea.
可能有看官会问预聚合不能解决细粒度窗口的问题吗答案是不能。预聚合只是让AggregateFunction/ReduceFunction之后的数据量降低但是进入WindowOperator的窗口状态的数据还是没变的。换句话说就算触发器实现为FIRE_AND_PURGE遍历大量窗口并写入状态的开销也是无法消除的。
扯了这么多有解决方案吗
当然是有的办法总比困难多。我们一般使用 滚动窗口在线存储读时聚合 的思路作为workaround。简单来讲就是
弃用滑动窗口用长度等于原滑动窗口步长的滚动窗口代替 每个滚动窗口将其周期内的数据做聚合打入外部在线存储内存数据库如RedisLSM-based NoSQL存储如HBase 扫描在线存储中对应时间区间可以灵活指定的所有行并将计算结果返回给前端展示。 针对上面的PV/UV问题如果采用Redis作为在线存储我们可以将时间戳放在key内并设定24小时过期时间。用数字字符串存储3分钟周期内的PV量用HyperLogLog存储3分钟周期内的UV量。近24小时的PV和UV就分别可以通过简单加减和HyperLogLog的pfmerge/pfcount命令得出了。当然实际操作起来还是要根据需求和服务器资源而定。 喵呜面试助手一站式解决面试问题你可以搜索微信小程序 [喵呜面试助手] 或关注 [喵呜刷题] - 面试助手 免费刷题。如有好的面试知识或技巧期待您的共享
