Flink之窗口触发机制及自定义Trigger的使用-Toy模板网

这篇具有很好参考价值的文章主要介绍了Flink之窗口触发机制及自定义Trigger的使用。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1 窗口触发机制

窗口计算的触发机制都是由Trigger类决定的,Flink中为各类内置的WindowsAssigner都设计了对应的默认Trigger. 层次结构如下:

Trigger
ProcessingTimeoutTrigger
EventTimeTrigger
CountTrigger
DeltaTrigger
NeverTrigger in GlobalWindows
ContinuousEventTimeTrigger
PurgingTrigger
ContinuousProcessingTimeTrigger
ProcessingTimeTrigger

通常情况下是不需要自己重写Trigger的,使用Flink内置的就可以,除非特殊业务特殊需求.

1.1 源码解析

以EventTimeTrigger源码说明如何触发窗口计算,在EventTimeTrigger源码中只需要关注onElement和onEventTime两个方法即可,源码内容如下:

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

    private EventTimeTrigger() {}

    // 基于数据驱动的方法
    @Override
    public TriggerResult onElement(
            Object element, long timestamp, TimeWindow window, TriggerContext ctx)
            throws Exception {
        // 判断当前watermark是否大于等于窗口的最大时间
        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;
        }
    }

    // 基于事件时间定时器驱动的方法
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
        // 根据不断发送来的watermark判断是否触发计算
        return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
    }
    // ...
}

源码中将不需要的关注的代码都已省略

onElement

注释中写明这个方法是基于数据进行驱动的,也就是说只有数据到达时才会执行这个方法,每一个窗口都有自己的startTime和endTime,也就是窗口的范围,判断条件中window.maxTimestamp()获取的就是当前窗口的endTime,如果当前watermark超出当前窗口的endTime就会触发这个窗口计算,TriggerResult.FIRE表示的就是窗口开始计算,如果当前watermark小于endTime就不会触发窗口计算这个窗口会继续等待数据输入,也就是TriggerResult.CONTINUE方法.
onEventTime

onElement是由数据驱动的,但是Flink的实际数据处理过程是存在没有数据发送到当前窗口,但是会有watermark源源不断的发送到当前窗口的情况,在多并行度的执行条件下就会发生这种情况.在onEventTime方法中如果上游发送过来的watermark等于当前窗口的endTime就会执行TriggerResult.FIRE否则还是执行TriggerResult.CONTINUE.

Trigger的触发机制就是这样,其他的CountTrigger等大致逻辑基本是一样的,了解清楚源码中这两个方法的作用很容易理解.

1.2 代码实现

通常Flink内置的Trigger都可以满足数据处理需求,往往在实际开发中可能会存在特殊的业务需求,这时用户可以自定义Trigger,以达到控制窗口触发计算的规则. 可以仿照EventTimeTrigger来构建一个自定义Trigger,只需要将其中的部分代码简单进行修改,并在onElement方法中添加自定的触发逻辑即可.

自定义Trigger

/**
 * 这里首先需要继承Trigger类,并将<Object, TimeWindow>中的Object修改成自己需要的数据类型,这段代码中需要根据UserEvent2中的数据
 * 来控制触发窗口计算的条件,所以将Object修改成UserEvent2
 **/ 
public class CustomTrigger extends Trigger<UserEvent2, TimeWindow> {
    public CustomTrigger() {}

    // 通过修改onElement方法中窗口计算的触发逻辑实现自定义方式
    @Override
    public TriggerResult onElement(
            // 这里也要将原有的Object类型修改成上面的UserEvent2
            UserEvent2 element, long timestamp, TimeWindow window, TriggerContext ctx)
            throws Exception {
        // 原有的判断逻辑不动,这个是为了便捷,判断逻辑可以根据实际需求进行修改,或者如同下面中添加一个新的触发逻辑
        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
            return TriggerResult.FIRE;
        // 这里增加一个判断逻辑,当用户行为时间为2700的时候也触发计算
        } else if (element.getTime().equals("2700")) {
            return TriggerResult.FIRE;
        // 原有的判断逻辑不动
        } else {
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }
    }

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

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

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        ctx.deleteEventTimeTimer(window.maxTimestamp());
    }

    @Override
    public boolean canMerge() {
        return true;
    }

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

    // 将toString中俄返回值根据用户的需要进行修改
    @Override
    public String toString() {
        return "CustomTrigger()";
    }

    // 将返回值更改成创建的自定义Trigger类
    public static CustomTrigger create() {
        return new CustomTrigger();
    }
}

业务代码
```
// ...
SingleOutputStreamOperator<UserEvent2> windowedStream = keyedStream
        .window(TumblingEventTimeWindows.of(Time.seconds(10))) // 设置滚动窗口大小为10s
        .trigger(new CustomTrigger()) // 传入自定义的Trigger类
        .allowedLateness(Time.seconds(2)) // 允许迟到数据迟到时间2s,同watermark中的forBoundedOutOfOrderness功能类似
        .sideOutputLateData(lateData) // 将迟到数据进行测流输出
        .max("time");// 获取用户行为发生事件最大的这条数据
// ...
```
上面这段业务代码中设置的滚动窗口的大小为10s,正常来说只有满足end - start = 10000的时候才会触发窗口计算,但是在自定义Trigger中指定了当数据中时间为2700的时候也触发窗口计算,在时间为2700的数据没到达时候还会按照原有的逻辑触发窗口计算,但是只要2700的数据到达,不管时候达到TumblingEventTimeWindows.of(Time.seconds(10))这个条件,都会触发窗口计算.文章来源地址https://www.toymoban.com/news/detail-854157.html