Flink触发器Trigger

🕗 发布于 2024-10-17 01:11 flink python 大数据

前言

在 Flink 窗口计算模型中，数据先经过 WindowAssigner 分配窗口，然后再经过触发器 Trigger，Trigger 决定了一个窗口何时被 ProcessFunction 处理。每个 WindowAssigner 都有一个默认的 Trigger，如果默认的不满足需求，可以通过 WindowedStream.trigger() 指定自定义的 Trigger。

认识Trigger

所有触发器都是org.apache.flink.streaming.api.windowing.triggers.Trigger的子类，父类定义了一个触发器应该具备的能力：

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

    public Trigger() {
    }

    public abstract TriggerResult onElement(T var1, long var2, W var4, TriggerContext var5) throws Exception;

    public abstract TriggerResult onProcessingTime(long var1, W var3, TriggerContext var4) throws Exception;

    public abstract TriggerResult onEventTime(long var1, W var3, TriggerContext var4) throws Exception;

    public boolean canMerge() {
        return false;
    }

    public void onMerge(W window, OnMergeContext ctx) throws Exception {
        throw new UnsupportedOperationException("This trigger does not support merging.");
    }

    public abstract void clear(W var1, TriggerContext var2) throws Exception;
}

Trigger 抽象类提供了六个方法：

onElement 元素被加入到窗口时触发，返回值决定窗口是否计算
onProcessingTime 注册的ProcessingTime任务时间到达时触发
onEventTime 注册的EventTime任务时间到达时触发
canMerge 是否可以合并
onMerge Trigger合并
clear 窗口被移除时触发

如果数据本身携带窗口是否触发计算的标记，那么重写 onElement() 方法即可；但是在时间窗口计算模型下，并不是通过元素来判断窗口是否需要计算的，而是窗口的结束时间到达时才出发计算，这个时候就需要用到定时器 Timer。Timer 就像一个闹钟，我们可以在它上面注册一个未来的时间戳，当这个时间到达时，对应的事件就会被触发，就像闹钟喊醒沉睡的你一样。

Trigger 会有自己的 Timer，TriggerContext 提供了注册时间事件的方法，你可以根据自己采用的时间语义，调用对应的注册方法来注册事件。

triggerContext.registerProcessingTimeTimer();
triggerContext.registerEventTimeTimer();

在重写这三个方法时，要重点关注方法的返回值。方法的返回值有两个作用：1、窗口内的数据是否可以计算，2、窗口内的数据是否需要清理。

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 枚举有四个值，含义分别是：

CONTINUE 不做任何操作
FIRE 触发窗口计算，但是数据仍然保留
PURGE 不触发计算，只是清理窗口内数据
FIRE_AND_PURGE 触发窗口计算，同时清理数据

内置的Trigger

Flink 内置了许多常用的 Trigger，大多数情况下它们足以支撑我们的业务场景，只有当内置的Trigger不符合要求时，才需要开发自定义的 Trigger。

1、EventTimeTrigger

和事件时间窗口搭配使用的 Trigger，直到 Watermark 时间戳等于窗口的结束时间才会触发计算。

onElement 的逻辑是：当有元素加入到窗口时，先判断当前 Watermark 时间戳是否到达窗口的结束时间，如果到了就直接触发计算，否则注册一个时间事件，等待 Timer 触发窗口计算。

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

    private EventTimeTrigger() {
    }

    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
            return TriggerResult.FIRE;
        } else {
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }
    }

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

2、ProcessingTimeTrigger

和 EventTimeTrigger 差不多，区别是采用的处理时间语义，没有 Watermark 相关的判断，直接注册ProcessingTime 事件等待窗口触发计算。

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

    private ProcessingTimeTrigger() {
    }

    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) {
        ctx.registerProcessingTimeTimer(window.maxTimestamp());
        return TriggerResult.CONTINUE;
    }

    public TriggerResult onProcessingTime(long time, TimeWindow window, Trigger.TriggerContext ctx) {
        return TriggerResult.FIRE;
    }
}

3、DeltaTrigger

DeltaTrigger 会计算新加入窗口的元素和上一个元素的差值，当这个差值超过给定的阈值时，窗口就会触发计算，元素之间的差值是通过指定的 DeltaFunction 计算出来的。

public class DeltaTrigger<T, W extends Window> extends Trigger<T, W> {
    private static final long serialVersionUID = 1L;
    private final DeltaFunction<T> deltaFunction;
    private final double threshold;
    private final ValueStateDescriptor<T> stateDesc;

    private DeltaTrigger(double threshold, DeltaFunction<T> deltaFunction, TypeSerializer<T> stateSerializer) {
        this.deltaFunction = deltaFunction;
        this.threshold = threshold;
        this.stateDesc = new ValueStateDescriptor("last-element", stateSerializer);
    }

    public TriggerResult onElement(T element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
        ValueState<T> lastElementState = (ValueState)ctx.getPartitionedState(this.stateDesc);
        if (lastElementState.value() == null) {
            lastElementState.update(element);
            return TriggerResult.CONTINUE;
        } else if (this.deltaFunction.getDelta(lastElementState.value(), element) > this.threshold) {
            lastElementState.update(element);
            return TriggerResult.FIRE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }
}

自定义Trigger

通过子类继承 Trigger 重写相应的方法，即可自定义我们自己的触发器。

举个例子，我们自定义一个和时间不相关的 Trigger，我们等窗口积攒到一定数量的元素再出发计算。如下示例程序：

public static class CounterTrigger extends Trigger<Long, GlobalWindow> {
        private final int count;

        public CounterTrigger(int count) {
            this.count = count;
        }

        @Override
        public TriggerResult onElement(Long element, long timestamp, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
            // 通过Flink state来保存窗口内积攒的元素数量
            ValueState<Integer> countState = triggerContext.getPartitionedState(new ValueStateDescriptor<Integer>("count", Integer.class));
            int elementCount = Optional.ofNullable(countState.value()).orElse(0) + 1;
            if (elementCount >= this.count) {
                countState.update(0);
                return TriggerResult.FIRE_AND_PURGE;
            }
            countState.update(elementCount);
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onProcessingTime(long timestamp, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
            return null;
        }

        @Override
        public TriggerResult onEventTime(long timestamp, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
            return null;
        }

        @Override
        public void clear(GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {

        }
    }

接下来验证一下我们的Trigger是否生效。

我们编写一个Flink作业，数据源每秒生成10个随机数，数据会被统一划分到 GlobalWindow 窗口，然后指定我们自定义的 CounterTrigger，等窗口内积攒了十条数据就出发求和计算。

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
    environment.addSource(new SourceFunction<Long>() {
                @Override
                public void run(SourceContext<Long> sourceContext) throws Exception {
                    while (true) {
                        Threads.sleep(100);
                        sourceContext.collect(ThreadLocalRandom.current().nextLong(100));
                    }
                }

                @Override
                public void cancel() {

                }
            }).windowAll(GlobalWindows.create())
            .trigger(new CounterTrigger(10))
            .process(new ProcessAllWindowFunction<Long, Object, GlobalWindow>() {
                @Override
                public void process(ProcessAllWindowFunction<Long, Object, GlobalWindow>.Context context, Iterable<Long> iterable, Collector<Object> collector) throws Exception {
                    Iterator<Long> iterator = iterable.iterator();
                    long sum = 0L;
                    while (iterator.hasNext()) {
                        sum += iterator.next();
                    }
                    System.err.println(sum);
                }
            });
    environment.execute();
}

运行Flink作业，控制台每隔一段时间就会输出随机数之和。

尾巴

在 Flink 中，Trigger 决定了何时触发窗口的计算和输出结果。通过灵活配置 Trigger 规则，能够精确控制数据处理的时机，适应不同的业务需求和数据特点。

Trigger 能够处理各种复杂的情况，例如在特定条件满足时触发，或者基于时间间隔、数据量等因素进行触发。它为开发者提供了精细的控制手段，确保数据处理的准确性和及时性。合理运用 Trigger 可以优化 Flink 作业的性能，避免不必要的计算和资源消耗。

原文地址：https://blog.csdn.net/qq_32099833/article/details/142958328

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Python画笔案例-085 绘制 3D效果文字
下一篇：重学SpringBoot3-集成Spring Boot Actuator

阿斯伯格与自闭症：特征对比及教育启示
在运动方面，自闭症孩子也可能存在各种问题，有的孩子动作不协调，有的孩子则可能过度活跃或过于安静。对于阿斯伯格综合征的孩子，教育者可以充分利用他们的兴趣爱好，引导他们在感兴趣的领域深入学习，培养他们的特
阅读更多2024-10-17
基于Java微信小程序的水果销售系统详细设计和实现(源码+lw+部署文档+讲解等)
请联系我获取更详细的演示视频。
阅读更多2024-10-17
MPA-SVM多变量回归预测|海洋捕食者优化算法-支持向量机|Matalb
本代码基于。
阅读更多2024-10-17
零基础学大模型——大模型技术学习过程梳理
从大的方向上来说，大模型从技术到应用，主要涉及到以上几个大的模块；而每个模块又涉及到大量的技术和细节。比如打造不同任务的神经网络模型，强化学习，迁移学习，知识蒸馏，分布式训练与存储等；以及RAG使用的
阅读更多2024-10-17
[C++] 小游戏征伐 SLG DNF 0.0.3 版本 zty出品
OK啊，废话不多说先赞后看养成习惯。
阅读更多2024-10-17
Linux——进程
冯·诺依曼结构作为现代计算机发展所遵循的结构，其主要思想包括：①将计算机分为了五个基本部件：运算器：进行四种基本算术运算和一些逻辑运算等；控制器：协调和控制计算机中各个组件的工作；存储器：存放数据与指
阅读更多2024-10-17
VLAN虚拟技术
复习：路由器的工作原理：根据路由表转发数据路由表的形成：自动获取1.直连路由2.动态路由 rip ospf静态获取手动配置网关配置：ip地址：1-223子网掩码；网关冲突域：集线器（已淘汰）：所有端口
阅读更多2024-10-17
深度学习：模型攻击（Model Attack）详解
模型攻击展示了当今AI系统可能面临的安全威胁。为了确保这些系统的安全和可靠，开发和部署AI解决方案时必须考虑到这些潜在的攻击，并采取适当的防御措施。通过综合安全性设计、持续监控和适时更新，可以显著降低
阅读更多2024-10-17
02.04、分割链表
使用ListNode()创建虚拟头节点，避免处理头节点的特殊情况，简化代码逻辑。遍历原链表的过程中，将节点分别加入到less或greater链表。遍历完后，将less链表与greater链表拼接在一起
阅读更多2024-10-17
Python知识点：基于Python技术，如何使用AirSim进行无人机模拟
AirSim是由微软开发的一个开源模拟器，它支持多种无人机模型，并且可以模拟真实的物理环境和传感器数据。AirSim提供了Python API，使得用户可以通过Python脚本控制无人机的行为，包括起
阅读更多2024-10-17

Flink触发器Trigger

前言

认识Trigger

内置的Trigger

自定义Trigger

尾巴

相关文章