Flink trigger

2021-07-24

Flink trigger简单介绍和部分源码解析

先从官网上拉下来trigger的介绍。

Triggers #

A Trigger determines when a window (as formed by the window assigner) is ready to be processed by the window function. Each WindowAssigner comes with a default Trigger. If the default trigger does not fit your needs, you can specify a custom trigger using trigger(...).

The trigger interface has five methods that allow a Trigger to react to different events:

The onElement() method is called for each element that is added to a window.
The onEventTime() method is called when a registered event-time timer fires.
The onProcessingTime() method is called when a registered processing-time timer fires.
The onMerge() method is relevant for stateful triggers and merges the states of two triggers when their corresponding windows merge, e.g. when using session windows.
Finally the clear() method performs any action needed upon removal of the corresponding window.

Two things to notice about the above methods are:

The first three decide how to act on their invocation event by returning a TriggerResult. The action can be one of the following:

CONTINUE: do nothing,
FIRE: trigger the computation,
PURGE: clear the elements in the window, and
FIRE_AND_PURGE: trigger the computation and clear the elements in the window afterwards.

Any of these methods can be used to register processing- or event-time timers for future actions.

Fire and Purge #

Once a trigger determines that a window is ready for processing, it fires, i.e., it returns FIRE or FIRE_AND_PURGE. This is the signal for the window operator to emit the result of the current window. Given a window with a ProcessWindowFunction all elements are passed to the ProcessWindowFunction (possibly after passing them to an evictor). Windows with ReduceFunction, or AggregateFunction simply emit their eagerly aggregated result.

When a trigger fires, it can either FIRE or FIRE_AND_PURGE. While FIRE keeps the contents of the window, FIRE_AND_PURGE removes its content. By default, the pre-implemented triggers simply FIRE without purging the window state.

Purging will simply remove the contents of the window and will leave any potential meta-information about the window and any trigger state intact.

Default Triggers of WindowAssigners #

The default Trigger of a WindowAssigner is appropriate for many use cases. For example, all the event-time window assigners have an EventTimeTrigger as default trigger. This trigger simply fires once the watermark passes the end of a window.

The default trigger of the GlobalWindow is the NeverTrigger which does never fire. Consequently, you always have to define a custom trigger when using a GlobalWindow.

By specifying a trigger using trigger() you are overwriting the default trigger of a WindowAssigner. For example, if you specify a CountTrigger for TumblingEventTimeWindows you will no longer get window firings based on the progress of time but only by count. Right now, you have to write your own custom trigger if you want to react based on both time and count.

Built-in and Custom Triggers #

Flink comes with a few built-in triggers.

The (already mentioned) EventTimeTrigger fires based on the progress of event-time as measured by watermarks.
The ProcessingTimeTrigger fires based on processing time.
The CountTrigger fires once the number of elements in a window exceeds the given limit.
The PurgingTrigger takes as argument another trigger and transforms it into a purging one.

If you need to implement a custom trigger, you should check out the abstract Trigger class. Please note that the API is still evolving and might change in future versions of Flink.

源码

结合源码简单分析一下。

@PublicEvolving
public abstract class Trigger<T, W extends Window> implements Serializable {
    private static final long serialVersionUID = -4104633972991191369L;

    public Trigger() {
    }

  // The onElement() method is called for each element that is added to a window.
    public abstract TriggerResult onElement(T var1, long var2, W var4, Trigger.TriggerContext var5) throws Exception;

  // The onProcessingTime() method is called when a registered processing-time timer fires.
    public abstract TriggerResult onProcessingTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception;

  // The onEventTime() method is called when a registered event-time timer fires.
    public abstract TriggerResult onEventTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception;

    public boolean canMerge() {
        return false;
    }

  // The onMerge() method is relevant for stateful triggers and merges the states of two triggers when their corresponding windows merge, e.g. when using session windows.
    public void onMerge(W window, Trigger.OnMergeContext ctx) throws Exception {
        throw new UnsupportedOperationException("This trigger does not support merging.");
    }

  // Finally the clear() method performs any action needed upon removal of the corresponding window.
    public abstract void clear(W var1, Trigger.TriggerContext var2) throws Exception;

    public interface OnMergeContext extends Trigger.TriggerContext {
        <S extends MergingState<?, ?>> void mergePartitionedState(StateDescriptor<S, ?> var1);
    }
}

结合官方文档的介绍，可以很清楚的了解每个方法的作用，可以看到主要执行方法都会返回一个元素叫TriggerResult。

public enum TriggerResult {
  // 有两个boolean属性，前一个代表是否计算，后一个代表是否清除
    CONTINUE(false, false),
    FIRE_AND_PURGE(true, true),
    FIRE(true, false),
    PURGE(false, true);

    private final boolean fire;
    private final boolean purge;

    private TriggerResult(boolean fire, boolean purge) {
        this.purge = purge;
        this.fire = fire;
    }

    public boolean isFire() {
        return this.fire;
    }

    public boolean isPurge() {
        return this.purge;
    }
}

值得一提的是，trigger的种类有很多，如下图。

我们看一下EventTimeTrigger源码

// extends Trigger
public class EventTimeTrigger extends Trigger<Object, TimeWindow> {
    private static final long serialVersionUID = 1L;

    private EventTimeTrigger() {
    }

    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
          // if the watermark is already past the window, fire immediately
            return TriggerResult.FIRE;
        } else {
          // 设置定时器
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }
    }

    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
        return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
    }

    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
      // 删除定时器
        ctx.deleteEventTimeTimer(window.maxTimestamp());
    }

    public boolean canMerge() {
        return true;
    }

    public void onMerge(TimeWindow window, OnMergeContext ctx) {
        long windowMaxTimestamp = window.maxTimestamp();
        if (windowMaxTimestamp > ctx.getCurrentWatermark()) {
            ctx.registerEventTimeTimer(windowMaxTimestamp);
        }

    }

    public String toString() {
        return "EventTimeTrigger()";
    }

    public static EventTimeTrigger create() {
        return new EventTimeTrigger();
    }
}

具体的使用例子在另一篇文章有写

链接

本文作者： xiaoyuyu
本文链接： http://woaixiaoyuyu.github.io/2021/07/24/Flink%20trigger/
版权声明： 本作品采用知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议进行许可。转载请注明出处！