微信商城网站怎么开发,九江有没有做网站的公司,关键词优化哪个好,长沙人才市场招聘信息一、触发器(Trigger)
Trigger 决定了一个窗口#xff08;由 window assigner 定义#xff09;何时可以被 window function 处理。 每个 WindowAssigner 都有一个默认的 Trigger。 如果默认 trigger 无法满足你的需要#xff0c;你可以在 trigger(…) 调用中指定自定义的 tr…一、触发器(Trigger)
Trigger 决定了一个窗口由 window assigner 定义何时可以被 window function 处理。 每个 WindowAssigner 都有一个默认的 Trigger。 如果默认 trigger 无法满足你的需要你可以在 trigger(…) 调用中指定自定义的 trigger。
1.1 Flink中预置的Trigger
窗口的计算触发依赖于窗口触发器每种类型的窗口都有对应的窗口触发机制都有一个默认的窗口触发器触发器的作用就是去控制什么时候来触发计算。flink内部定义多种触发器每种触发器对应于不同的WindowAssigner。常见的触发器如下
EventTimeTrigger通过对比EventTime和窗口的Endtime确定是否触发窗口计算如果EventTime大于Window EndTime则触发否则不触发窗口将继续等待。ProcessTimeTrigger通过对比ProcessTime和窗口EndTme确定是否触发窗口如果ProcessTime大于EndTime则触发计算否则窗口继续等待。ProcessingTimeoutTrigger可以将任何触发器转变为超时触发器。ContinuousEventTimeTrigger根据间隔时间周期性触发窗口或者Window的结束时间小于当前EndTime触发窗口计算。ContinuousProcessingTimeTrigger根据间隔时间周期性触发窗口或者Window的结束时间小于当前ProcessTime触发窗口计算。CountTrigger根据接入数据量是否超过设定的阙值判断是否触发窗口计算。DeltaTrigger根据接入数据计算出来的Delta指标是否超过指定的Threshold去判断是否触发窗口计算。PurgingTrigger可以将任意触发器作为参数转换为Purge类型的触发器计算完成后数据将被清理。NeverTrigger任何时候都不触发窗口计算
1.2 Trigger的抽象类
Trigger 接口提供了五个方法来响应不同的事件
onElement() 方法在每个元素被加入窗口时调用。onEventTime() 方法在注册的 event-time timer 触发时调用。onProcessingTime() 方法在注册的 processing-time timer 触发时调用。canMerge() 方法判断是否可以合并。onMerge() 方法与有状态的 trigger 相关。该方法会在两个窗口合并时 将窗口对应 trigger 的状态进行合并比如使用会话窗口时。clear() 方法处理在对应窗口被移除时所需的逻辑。
触发器接口的源码如下
PublicEvolving
public abstract class TriggerT, W extends Window implements Serializable {private static final long serialVersionUID -4104633972991191369L;/*** Called for every element that gets added to a pane. The result of this will determine whether* the pane is evaluated to emit results.** param element The element that arrived.* param timestamp The timestamp of the element that arrived.* param window The window to which the element is being added.* param ctx A context object that can be used to register timer callbacks.*/public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx)throws Exception;/*** Called when a processing-time timer that was set using the trigger context fires.** param time The timestamp at which the timer fired.* param window The window for which the timer fired.* param ctx A context object that can be used to register timer callbacks.*/public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx)throws Exception;/*** Called when an event-time timer that was set using the trigger context fires.** param time The timestamp at which the timer fired.* param window The window for which the timer fired.* param ctx A context object that can be used to register timer callbacks.*/public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx)throws Exception;/*** Returns true if this trigger supports merging of trigger state and can therefore be used with* a {link org.apache.flink.streaming.api.windowing.assigners.MergingWindowAssigner}.** pIf this returns {code true} you must properly implement {link #onMerge(Window,* OnMergeContext)}*/public boolean canMerge() {return false;}/*** Called when several windows have been merged into one window by the {link* org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}.** param window The new window that results from the merge.* param ctx A context object that can be used to register timer callbacks and access state.*/public void onMerge(W window, OnMergeContext ctx) throws Exception {throw new UnsupportedOperationException(This trigger does not support merging.);}/*** Clears any state that the trigger might still hold for the given window. This is called when* a window is purged. Timers set using {link TriggerContext#registerEventTimeTimer(long)} and* {link TriggerContext#registerProcessingTimeTimer(long)} should be deleted here as well as* state acquired using {link TriggerContext#getPartitionedState(StateDescriptor)}.*/public abstract void clear(W window, TriggerContext ctx) throws Exception;// ------------------------------------------------------------------------/*** A context object that is given to {link Trigger} methods to allow them to register timer* callbacks and deal with state.*/public interface TriggerContext {// ...}/*** Extension of {link TriggerContext} that is given to {link Trigger#onMerge(Window,* OnMergeContext)}.*/public interface OnMergeContext extends TriggerContext {S extends MergingState?, ? void mergePartitionedState(StateDescriptorS, ? stateDescriptor);}
}关于上述方法需要注意三件事
1前三个方法返回TriggerResult枚举类型其包含四个枚举值
CONTINUE表示对窗口不执行任何操作。即不触发窗口计算也不删除元素。FIRE触发窗口计算但是保留窗口元素。PURGE不触发窗口计算丢弃窗口并且删除窗口的元素。FIRE_AND_PURGE触发窗口计算输出结果然后将窗口中的数据和窗口进行清除。
源码如下
public enum TriggerResult {// 不触发也不删除元素CONTINUE(false, false),// 触发窗口窗口出发后删除窗口中的元素FIRE_AND_PURGE(true, true),// 触发窗口但是保留窗口元素FIRE(true, false),// 不触发窗口丢弃窗口并且删除窗口的元素PURGE(false, true);// ------------------------------------------------------------------------private final boolean fire;private final boolean purge;TriggerResult(boolean fire, boolean purge) {this.purge purge;this.fire fire;}public boolean isFire() {return fire;}public boolean isPurge() {return purge;}
}(2) 每一个窗口分配器都拥有一个属于自己的 TriggerTrigger上会有定时器用来决定一个窗口何时能够被计算或清除当定时器触发后会调用对应的回调返回返回TriggerResult。Trigger的返回结果可以是 continue不做任何操作fire处理窗口数据purge移除窗口和窗口中的数据或者 fire purge。一个Trigger的调用结果只是fire的话那么会计算窗口并保留窗口原样也就是说窗口中的数据仍然保留不变等待下次Trigger fire的时候再次执行计算。一个窗口可以被重复计算多次知道它被 purge 了。在purge之前窗口会一直占用着内存。
1.3 ProcessingTimeTrigger源码分析
PublicEvolving
public class ProcessingTimeTrigger extends TriggerObject, TimeWindow {private static final long serialVersionUID 1L;private ProcessingTimeTrigger() {}Overridepublic TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {ctx.registerProcessingTimeTimer(window.maxTimestamp());return TriggerResult.CONTINUE;}Overridepublic TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx)throws Exception {return TriggerResult.CONTINUE;}Overridepublic TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {return TriggerResult.FIRE;}Overridepublic void clear(TimeWindow window, TriggerContext ctx) throws Exception {ctx.deleteProcessingTimeTimer(window.maxTimestamp());}Overridepublic boolean canMerge() {return true;}Overridepublic void onMerge(TimeWindow window, OnMergeContext ctx) {// only register a timer if the time is not yet past the end of the merged window// this is in line with the logic in onElement(). If the time is past the end of// the window onElement() will fire and setting a timer here would fire the window twice.long windowMaxTimestamp window.maxTimestamp();if (windowMaxTimestamp ctx.getCurrentProcessingTime()) {ctx.registerProcessingTimeTimer(windowMaxTimestamp);}}Overridepublic String toString() {return ProcessingTimeTrigger();}/** Creates a new trigger that fires once system time passes the end of the window. */public static ProcessingTimeTrigger create() {return new ProcessingTimeTrigger();}
}在 onElement()方法中ctx.registerProcessingTimeTimer(window.maxTimestamp())将会注册一个ProcessingTime定时器时间参数是window.maxTimestamp()也就是窗口的最终时间当时间到达这个窗口最终时间定时器触发并调用 onProcessingTime()方法在 onProcessingTime() 方法中return TriggerResult.FIRE 即返回 FIRE触发窗口中数据的计算但是会保留窗口元素。
需要注意的是ProcessingTimeTrigger类只会在窗口的最终时间到达的时候触发窗口函数的计算计算完成后并不会清除窗口中的数据这些数据存储在内存中除非调用PURGE或FIRE_AND_PURGE否则数据将一直存在内存中。实际上Flink中提供的Trigger类除了PurgingTrigger类其他的都不会对窗口中的数据进行清除。
EventTimeTriggerr在onElement设置的定时器 EventTime通过registerEventTimeTimer注册定时器在内部Watermark达到或超过Timer设定的时间戳时触发。
二、移除器(Evictor)
2.1 Evictor扮演的角色 当一个元素进入stream中之后一般要经历Window开窗、Trigger触发器、Evitor移除器、Windowfunction窗口计算操作具体过程如下
Window中的WindowAssigner窗口分配器定义了数据应该被分配到哪个窗口中每一个 WindowAssigner都会有一个默认的Trigger如果用户在代码中指定了窗口的trigger默认的 trigger 将会被覆盖。Trigger上会有定时器用来决定一个窗口何时能够被计算或清除。Trigger的返回结果可以是 continue不做任何操作fire处理窗口数据purge移除窗口和窗口中的数据或者 fire purge。一个Trigger的调用结果只是fire的话那么会计算窗口并保留窗口原样也就是说窗口中的数据仍然保留不变等待下次Trigger fire的时候再次执行计算。一个窗口可以被重复计算多次知道它被 purge 了。在purge之前窗口会一直占用着内存。当Trigger fire了窗口中的元素集合就会交给Evictor如果指定了的话。Evictor 主要用来遍历窗口中的元素列表并决定最先进入窗口的多少个元素需要被移除。剩余的元素会交给用户指定的函数进行窗口的计算。如果没有 Evictor 的话窗口中的所有元素会一起交给WindowFunction进行计算。WindowFunction收到了窗口的元素可能经过了 Evictor 的过滤并计算出窗口的结果值并发送给下游。窗口计算操作有很多比如预定义的sum(),min(),max()还有 ReduceFunctionWindowFunction。WindowFunction 是最通用的计算函数其他的预定义的函数基本都是基于该函数实现的。
现在大致了解了Evitor移除器扮演的角色和移除器在流中的哪个位置让我们继续看为何使用Evictor。
Evictor接口定义如下 evictBefore()包含要在窗口函数之前应用的清除逻辑而evictAfter()包含要在窗口函数之后应用的清除逻辑。应用窗口函数之前清除的元素将不会被窗口函数处理。
窗格是具有相同Key和相同窗口的元素组成的桶即同一个窗口中相同Key的元素一定属于同一个窗格。一个元素可以在多个窗格中当一个元素被分配给多个窗口时这些窗格都有自己的清除器实例。 注window默认没有evictor一旦把window指定Evictor该window会由EvictWindowOperator类来负责操作。 2.2 Flink内置的Evitor
CountEvictor保留窗口中用户指定的元素数量并丢弃窗口缓冲区剩余的元素。DeltaEvictor依次计算窗口缓冲区中的最后一个元素与其余每个元素之间的delta值若delta值大于等于指定的阈值则该元素会被移除。使用DeltaEvictor清除器需要指定两个参数一个是double类型的阈值另一个是DeltaFunction接口的实例DeltaFunction用于指定具体的delta值计算逻辑。TimeEvictor传入一个以毫秒为单位的时间间隔参数例如以size表示对于给定的窗口取窗口中元素的最大时间戳例如以max表示使用TimeEvictor清除器将删除所有时间戳小于或等于max-size的元素即清除从窗口开头到指定的截止时间之间的元素。
2.2.1 CountEvictor
private void evict(IterableTimestampedValueObject elements, int size, EvictorContext ctx) {if (size maxCount) {// 小于最大数量不做处理return;} else {int evictedCount 0;for (IteratorTimestampedValueObject iterator elements.iterator(); iterator.hasNext();){iterator.next();evictedCount;if (evictedCount size - maxCount) {break;} else {// 移除前size - maxCount个元素只剩下最后maxCount个元素iterator.remove();}}}
}2.2.2 DeltaEvictor
DeltaEvictor通过计算DeltaFunction的值依次传入每个元素和最后一个元素并将其与threshold进行对比如果DeltaFunction计算结果大于等于threshold则该元素会被移除。DeltaEvictor的实现如下
private void evict(IterableTimestampedValueT elements, int size, EvictorContext ctx) {// 获取最后一个元素TimestampedValueT lastElement Iterables.getLast(elements);for (IteratorTimestampedValueT iterator elements.iterator(); iterator.hasNext();){TimestampedValueT element iterator.next();// 依次计算每个元素和最后一个元素的delta值同时和threshold的值进行比较// 若计算结果大于threshold值或者是相等则该元素会被移除if (deltaFunction.getDelta(element.getValue(), lastElement.getValue()) this.threshold) {iterator.remove();}}
}2.2.3 TimeEvictor
TimeEvictor以时间为判断标准决定元素是否会被移除。TimeEvictor会获取窗口中所有元素的最大时间戳currentTimecurrentTime减去窗口大小(windowSize) 可得到能保留最久的元素的时间戳evictCutoff然后再遍历窗口中的元素如果元素的时间戳小于evictCutoff就执行移除操作否则不移除。具体逻辑如下图所示 TimeEvictor的代码实现如下
private void evict(IterableTimestampedValueObject elements, int size, EvictorContext ctx) {// 如果element没有timestamp直接返回if (!hasTimestamp(elements)) {return;}// 获取elements中最大的时间戳到来最晚的元素的时间long currentTime getMaxTimestamp(elements);// 截止时间为 到来最晚的元素的时间 - 窗口大小可以理解为保留最近的多久的元素long evictCutoff currentTime - windowSize;for (IteratorTimestampedValueObject iterator elements.iterator(); iterator.hasNext(); ) {TimestampedValueObject record iterator.next();// 清除所有时间戳小于截止时间的元素if (record.getTimestamp() evictCutoff) {iterator.remove();}}
}
