首页 > 资讯 > 后端开发 > Python >JavaFlink窗口触发器Trigger的用法详解

445

分享到

JavaFlink窗口触发器Trigger的用法详解

2024-04-02 19:04:59 445人浏览独家记忆

Python 官方文档：入门教程 => 点击学习

摘要

目录定义Trigger 源码TriggerResult 源码flink 预置的TriggerEventTimeTrigger源码ProcessingTimeTrigger源码常见窗口

定义

Trigger确定窗口（由窗口分配器形成）何时准备好由窗口函数处理。每个WindowAssigner都带有一个默认值Trigger。如果默认触发器不符合您的需求，您可以使用trigger(…)。

Trigger 源码

public abstract class Trigger<T, W extends Window> implements Serializable {
	
    public abstract TriggerResult onElement(T var1, long var2, W var4, Trigger.TriggerContext var5) throws Exception;
	
	 
    public abstract TriggerResult onProcessingTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception;

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

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

TriggerResult 源码

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;

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

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

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

一旦触发器确定窗口已准备好进行处理，就会触发，返回状态可以是FIRE或FIRE_AND_PURGE。其中FIRE是触发窗口计算并保留窗口内容，而FIRE_AND_PURGE是触发窗口计算并删除窗口内容。默认情况下，预实现的触发器只是简单地FIRE不清除窗口状态。

Flink 预置的Trigger

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

主要看看EventTimeTrigger和ProcessingTimeTrigger的源码。

EventTimeTrigger源码

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()) {
            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();
    }
}

ProcessingTimeTrigger源码

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

    private ProcessingTimeTrigger() {
    }

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

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

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

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

    public boolean canMerge() {
        return true;
    }

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

    }

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

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

在 onElement()方法中，ctx.registerProcessingTimeTimer(window.maxTimestamp())将会注册一个ProcessingTime定时器，时间参数是window.maxTimestamp()，也就是窗口的最终时间，当时间到达这个窗口最终时间，定时器触发并调用 onProcessingTime()方法，在 onProcessingTime() 方法中，return TriggerResult.FIRE 即返回 FIRE，触发窗口中数据的计算，但是会保留窗口元素。

需要注意的是ProcessingTimeTrigger类只会在窗口的最终时间到达的时候触发窗口函数的计算，计算完成后并不会清除窗口中的数据，这些数据存储在内存中，除非调用PURGE或FIRE_AND_PURGE，否则数据将一直存在内存中。实际上，Flink中提供的Trigger类，除了PurgingTrigger类，其他的都不会对窗口中的数据进行清除。

常见窗口的Trigger

滚动窗口

TumblingEventTimewindows ：EventTimeTrigger
public class TumblingEventTimeWindows extends WindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
            return EventTimeTrigger.create();
        }
}

TumblingProcessingTimeWindows ：ProcessingTimeTrigger

public class TumblingProcessingTimeWindows extends WindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return ProcessingTimeTrigger.create();
    }
}

滑动窗口

SlidingEventTimeWindows：EventTimeTrigger
public class SlidingEventTimeWindows extends WindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return EventTimeTrigger.create();
    }
}

SlidingProcessingTimeWindows ：ProcessingTimeTrigger

public class SlidingProcessingTimeWindows extends WindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
            return ProcessingTimeTrigger.create();
        }
}

会话窗口

EventTimeSessionWindows：EventTimeTrigger
public class EventTimeSessionWindows extends MergingWindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return EventTimeTrigger.create();
    }
}

ProcessingTimeSessionWindows：ProcessingTimeTrigger

public class ProcessingTimeSessionWindows extends MergingWindowAssigner<Object, TimeWindow> {
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return ProcessingTimeTrigger.create();
    }
}

全局窗口

GlobalWindows ：NeverTrigger
public class GlobalWindows extends WindowAssigner<Object, GlobalWindow> {
     public Trigger<Object, GlobalWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
            return new GlobalWindows.NeverTrigger();
        }
}

到此这篇关于Java Flink窗口触发器Trigger的用法详解的文章就介绍到这了,更多相关Java Flink窗口触发器内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

点击免费下载>>软考高级考试备考技巧/历年真题/备考精华资料

--结束END--

本文标题: JavaFlink窗口触发器Trigger的用法详解

本文链接: https://www.lsjlt.com/news/153939.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

本篇文章演示代码以及资料文档资料下载

下载Word文档到电脑，方便收藏和打印～

下载Word文档

去做题

回答

如何调试操作系统的错误？
操作系统

2023-11-15发布

回答

操作系统中的I/O系统是如何实现的？
操作系统

2023-11-15发布

回答

如何实现操作系统的内存管理？
操作系统

2023-11-15发布

回答

什么是虚拟内存，它对操作系统有什么影响？
操作系统

2023-11-15发布

回答

ASP中的MVC架构和WebForms架构有什么区别和使用场景？
ASP.NET

2023-11-15发布

回答

ASP中的数据验证和数据校验有什么不同？
ASP.NET

2023-11-15发布

回答

ASP中的ADO对象和DAO对象有什么区别和使用方法？
ASP.NET

2023-11-15发布

回答

Node.js中的包管理器NPM是什么？如何使用它进行依赖管理？
node.js

2023-11-15发布

回答

Vue.js中的动态组件是什么？如何使用它来动态渲染组件？
VUE

2023-11-15发布

回答

如何使用Vue.js实现懒加载和预加载？
VUE

2023-11-15发布

JavaFlink窗口触发器Trigger的用法详解

目录

定义

Trigger 源码

TriggerResult 源码

Flink 预置的Trigger

EventTimeTrigger源码

ProcessingTimeTrigger源码

常见窗口的Trigger

滚动窗口

滑动窗口

会话窗口

全局窗口

本篇文章演示代码以及资料文档资料下载

JavaFlink窗口触发器Trigger的用法详解

Java Flink窗口触发器Trigger怎么使用

详细解读MySQL的触发器trigger

Oracle Trigger触发器的正确使用方法

SQLServer触发器调用JavaWeb接口的过程详解

MySQL触发器的使用详解

Java Quartz触发器CronTriggerBean配置用法详解

MySQL触发器基本用法详解【创建、查看、删除等】

oracle 存储过程、函数和触发器用法实例详解

解决Fedora14下eclipse进行android开发,ibus提示没有输入窗口的方法详解

Golang并发利器sync.Once的用法详解

SQLServer2005触发器提示其他会话正在使用事务的解决方法

谷歌浏览器F12开发者工具NETWORK的用法详解

详细讲解 MySQL的重要概念和用法，例如索引、事务、存储过程、触发器、视图、权限管理等，值得一看！

Py的itchat库：python开发神器之itchat的介绍、安装、使用方法详解

python分析数据的方法是什么

如何使用Python实现抽奖小程序

python copy函数的作用是什么

python进程池创建队列的方法是什么

python字符串处理与应用的方法有哪些

python全局变量如何定义

python如何读取文件夹下所有文件

python keyerror错误怎么解决

python如何提取字符串的数字

python中怎么将回车作为输入内容