org.apache.flink.streaming.api.windowing.triggers;

 

Trigger

public abstract class Trigger
     
       implements Serializable {    /**     * 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;    /**     * 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 TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {        throw new RuntimeException("This trigger does not support merging.");    }
     

Trigger决定pane何时被evaluated，实现一系列接口，来判断各种情况下是否需要trigger

看看具体的trigger的实现，

ProcessingTimeTrigger

/** * A {
    @link Trigger} that fires once the current system time passes the end of the window * to which a pane belongs. */public class ProcessingTimeTrigger implements Trigger
     
       {    private static final long serialVersionUID = 1L;        private ProcessingTimeTrigger() {}        @Override    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {        ctx.registerProcessingTimeTimer(window.maxTimestamp()); //对于processingTime，element的trigger时间是current+window，所以这里需要注册定时器去触发        return TriggerResult.CONTINUE;    }        @Override    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {        return TriggerResult.CONTINUE;    }        @Override    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {
    //触发后调用        return TriggerResult.FIRE_AND_PURGE;    }        @Override    public 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();    }}
     

可以看到只有在onProcessingTime的时候，是FIRE_AND_PURGE，其他时候都是continue

再看个CountTrigger，

public class CountTrigger
     
       extends Trigger
      
        {    private final long maxCount;    private final ReducingStateDescriptor
       
         stateDesc =            new ReducingStateDescriptor<>("count", new Sum(), LongSerializer.INSTANCE);    private CountTrigger(long maxCount) {        this.maxCount = maxCount;    }    @Override    public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {        ReducingState
        
          count = ctx.getPartitionedState(stateDesc); //从backend取出conunt state        count.add(1L); //加1        if (count.get() >= maxCount) {            count.clear();            return TriggerResult.FIRE;        }        return TriggerResult.CONTINUE;    }    @Override    public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {        return TriggerResult.CONTINUE;    }    @Override    public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {        return TriggerResult.CONTINUE;    }    @Override    public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {        ctx.mergePartitionedState(stateDesc); //先调用merge，底层backend里面的window进行merge        ReducingState
         
           count = ctx.getPartitionedState(stateDesc); //merge后再取出state，count，进行判断        if (count.get() >= maxCount) {            return TriggerResult.FIRE;        }        return TriggerResult.CONTINUE;    }
         
        
       
      
     

很简单，既然是算count，那么和time相关的自然都是continue

对于count，是在onElement中触发，每次来element都会走到这个逻辑

当累积的count > 设定的count时，就会返回Fire，注意，这里这是fire，并不会purge

并将计数清0

 

TriggerResult

TriggerResult是个枚举，

enum TriggerResult {    CONTINUE(false, false), FIRE_AND_PURGE(true, true), FIRE(true, false), PURGE(false, true);        private final boolean fire;    private final boolean purge;}

两个选项，fire，purge，2×2，所以4种可能性

两个Result可以merge，

/** * Merges two {
    @code TriggerResults}. This specifies what should happen if we have * two results from a Trigger, for example as a result from * {
    @link Trigger#onElement(Object, long, Window, Trigger.TriggerContext)} and * {
    @link Trigger#onEventTime(long, Window, Trigger.TriggerContext)}. * * 
 * For example, if one result says {
    @code CONTINUE} while the other says {
    @code FIRE} * then {
    @code FIRE} is the combined result; */public static TriggerResult merge(TriggerResult a, TriggerResult b) {    if (a.purge || b.purge) {        if (a.fire || b.fire) {            return FIRE_AND_PURGE;        } else {            return PURGE;        }    } else if (a.fire || b.fire) {        return FIRE;    } else {        return CONTINUE;    }}

 

TriggerContext

为Trigger做些环境的工作，比如管理timer，和处理state

这些接口在，Trigger中的接口逻辑里面都会用到，所以在Trigger的所有接口上，都需要传入context

/**     * A context object that is given to {
    @link Trigger} methods to allow them to register timer     * callbacks and deal with state.     */    public interface TriggerContext {        long getCurrentProcessingTime();        long getCurrentWatermark();            /**         * Register a system time callback. When the current system time passes the specified         * time {
    @link Trigger#onProcessingTime(long, Window, TriggerContext)} is called with the time specified here.         *         * @param time The time at which to invoke {
    @link Trigger#onProcessingTime(long, Window, TriggerContext)}         */        void registerProcessingTimeTimer(long time);        void registerEventTimeTimer(long time);            void deleteProcessingTimeTimer(long time);        void deleteEventTimeTimer(long time);             S getPartitionedState(StateDescriptor
      
        stateDescriptor);    }
      

 

OnMergeContext 仅仅是多了一个接口，

public interface OnMergeContext extends TriggerContext {    > void mergePartitionedState(StateDescriptor
      
        stateDescriptor);}
      

 

WindowOperator.Context作为TriggerContext的一个实现，

/** * {
    @code Context} is a utility for handling {
    @code Trigger} invocations. It can be reused * by setting the {
    @code key} and {
    @code window} fields. No internal state must be kept in * the {
    @code Context} */public class Context implements Trigger.OnMergeContext {    protected K key; //Context对应的window上下文    protected W window;    protected Collection
     
       mergedWindows; //onMerge中被赋值    @SuppressWarnings("unchecked")    public 
       S getPartitionedState(StateDescriptor
       
         stateDescriptor) {        try {            return WindowOperator.this.getPartitionedState(window, windowSerializer, stateDescriptor); //从backend里面读出改window的状态，即window buffer        } catch (Exception e) {            throw new RuntimeException("Could not retrieve state", e);        }    }    @Override    public 
        > void mergePartitionedState(StateDescriptor
         
           stateDescriptor) {        if (mergedWindows != null && mergedWindows.size() > 0) {            try {                WindowOperator.this.getStateBackend().mergePartitionedStates(window, //在backend层面把mergedWindows merge到window中                        mergedWindows,                        windowSerializer,                        stateDescriptor);            } catch (Exception e) {                throw new RuntimeException("Error while merging state.", e);            }        }    }    @Override    public void registerProcessingTimeTimer(long time) {        Timer
          
            timer = new Timer<>(time, key, window); // make sure we only put one timer per key into the queue if (processingTimeTimers.add(timer)) { processingTimeTimersQueue.add(timer); //If this is the first timer added for this timestamp register a TriggerTask if (processingTimeTimerTimestamps.add(time, 1) == 0) { //如果这个window是第一次注册的话 ScheduledFuture
            scheduledFuture = WindowOperator.this.registerTimer(time, WindowOperator.this); //对于processTime必须注册定时器主动触发 processingTimeTimerFutures.put(time, scheduledFuture); } } } @Override public void registerEventTimeTimer(long time) { Timer
           
             timer = new Timer<>(time, key, window); if (watermarkTimers.add(timer)) { watermarkTimersQueue.add(timer); } } //封装一遍trigger的接口，并把self作为context传入trigger的接口中 public TriggerResult onElement(StreamRecord
            
              element) throws Exception { return trigger.onElement(element.getValue(), element.getTimestamp(), window, this); } public TriggerResult onProcessingTime(long time) throws Exception { return trigger.onProcessingTime(time, window, this); } public TriggerResult onEventTime(long time) throws Exception { return trigger.onEventTime(time, window, this); } public TriggerResult onMerge(Collection
             
               mergedWindows) throws Exception { this.mergedWindows = mergedWindows; return trigger.onMerge(window, this); }}
             
            
           
          
         
       
     

 

 

Evictor

/** * An {
    @code Evictor} can remove elements from a pane before it is being processed and after * window evaluation was triggered by a * {
    @link org.apache.flink.streaming.api.windowing.triggers.Trigger}. * * 
 * A pane is the bucket of elements that have the same key (assigned by the * {
    @link org.apache.flink.api.java.functions.KeySelector}) and same {
    @link Window}. An element can * be in multiple panes of it was assigned to multiple windows by the * {
    @link org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}. These panes all * have their own instance of the {
    @code Evictor}. * * @param 
      
        The type of elements that this {
    @code Evictor} can evict. * @param 
       
         The type of {
    @link Window Windows} on which this {
    @code Evictor} can operate. */public interface Evictor
        
          extends Serializable {    /**     * Computes how many elements should be removed from the pane. The result specifies how     * many elements should be removed from the beginning.     *     * @param elements The elements currently in the pane.     * @param size The current number of elements in the pane.     * @param window The {
    @link Window}     */    int evict(Iterable
         
          
           > elements, int size, W window);}
          
         
        
       
      

Evictor的目的就是在Trigger fire后，但在element真正被处理前，从pane中remove掉一些数据

比如你虽然是每小时触发一次，但是只是想处理最后10分钟的数据，而不是所有数据。。。

 

CountEvictor

/** * An {
    @link Evictor} that keeps only a certain amount of elements. * * @param 
     
       The type of {
    @link Window Windows} on which this {
    @code Evictor} can operate. */public class CountEvictor
      
        implements Evictor
       
         {    private static final long serialVersionUID = 1L;        private final long maxCount;        private CountEvictor(long count) {        this.maxCount = count;    }        @Override    public int evict(Iterable
        
         
        
       
      
     

初始化count，表示想保留多少elements(from end)

evict返回需要删除的elements数目（from begining）

如果element数大于保留数，我们需要删除size – maxCount（from begining）

反之，就全保留

 

TimeEvictor

/** * An {
    @link Evictor} that keeps elements for a certain amount of time. Elements older * than {
    @code current_time - keep_time} are evicted. * * @param 
     
       The type of {
    @link Window Windows} on which this {
    @code Evictor} can operate. */public class TimeEvictor
      
        implements Evictor
       
         {    private static final long serialVersionUID = 1L;        private final long windowSize;        public TimeEvictor(long windowSize) {        this.windowSize = windowSize;    }        @Override    public int evict(Iterable
        
         
        
       
      
     

TimeEvictor设置需要保留的时间，

用最后一条的时间作为current，current-windowSize，作为界限，小于这个时间的要evict掉

这里的前提是，数据是时间有序的