Flink学习笔记-Trigger窗口触发

74 阅读 0 评论 49 点赞

我是靠谱客的博主细心长颈鹿，最近开发中收集的这篇文章主要介绍Flink学习笔记-Trigger窗口触发，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

Flink学习笔记-Trigger窗口触发

- Flink提供的Triggers
- 自定义Trigger

窗口的计算触发依赖于窗口触发器，每种类型的窗口都有对应的窗口触发机制，都有一个默认的窗口触发器，触发器的作用就是去控制什么时候来触发我们的聚合方法。Flink内部定义实现了如EventTimeTrigger、ProcessTimeTrigger、CountTrigger以及DeltaTrigger等等。一般地，每种触发器对应于不同的Window Assigner，例如EventTime类型的Windows对应的触发器是EventTimeTrigger，工作原理是判断当前的Watermark是否超过了窗口的EndTime，如果超过则触发对窗口内数据的计算，否则不触发计算。

Flink提供的Triggers

在这里插入图片描述

EventTimeTrigger：通过对比Watermark和窗口的Endtime确定是否触发窗口计算，如果Watermark大于Window EndTime则触发，否则不触发，窗口将继续等待。
ProcessTimeTrigger：通过对比ProcessTime和窗口EndTme确定是否触发窗口，如果ProcessTime大于EndTime则触发计算，否则窗口继续等待。
ContinuousEventTimeTrigger：根据间隔时间周期性触发窗口或者Window的结束时间小于当前EndTime触发窗口计算。
ContinuousProcessingTimeTrigger：根据间隔时间周期性触发窗口或者Window的结束时间小于当前ProcessTime触发窗口计算。
CountTrigger：根据接入数据量是否超过设定的阙值判断是否触发窗口计算。
DeltaTrigger：根据接入数据计算出来的Delta指标是否超过指定的Threshold去判断是否触发窗口计算。
PurgingTrigger：可以将任意触发器作为参数转换为Purge类型的触发器，计算完成后数据将被清理。

自定义Trigger

继承实现Trigger抽象类去完成自定义触发器，然后在DataStream中调用trigger方法传入自定义的Trigger即可。

package myLearn;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author qingh.yxb
 * @since 2019/8/10
 */
public class MyTrigger extends Trigger<Tuple2<String, Integer>, TimeWindow> {
    private static final long serialVersionUID = 1L;
    ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("total", Integer.class);

    @Override
    public TriggerResult onElement(Tuple2<String, Integer> element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        ValueState<Integer> sumState = ctx.getPartitionedState(stateDescriptor);
        if (null == sumState.value()) {
            sumState.update(0);
        }
        sumState.update(element.f1 + sumState.value());
        if (sumState.value() >= 2) {
            //这里可以选择手动处理状态
            //  默认的trigger发送是TriggerResult.FIRE 不会清除窗口数据
            return TriggerResult.FIRE_AND_PURGE;
        }
        return TriggerResult.CONTINUE;
    }

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

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

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        System.out.println("清理窗口状态  窗口内保存值为" + ctx.getPartitionedState(stateDescriptor).value());
        ctx.getPartitionedState(stateDescriptor).clear();
    }
}