44、Flink 的默认窗口剔除器 evictor 代码示例

2024-06-14 10:44
文章标签 代码 窗口 示例 默认 flink 44 剔除 evictor

本文主要是介绍44、Flink 的默认窗口剔除器 evictor 代码示例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1、CountEvictor

仅记录用户指定数量的元素，一旦窗口中的元素超过这个数量，多余的元素会从窗口缓存的开头移除。

2、DeltaEvictor

接收 DeltaFunction 和 threshold 参数，计算最后一个元素与窗口缓存中所有元素的差值，并移除差值大于或等于 threshold 的元素。

3、TimeEvictor

接收 interval 参数，以毫秒表示，它会找到窗口中元素的最大 timestamp max_ts，并移除比 max_ts - interval 小的所有元素。

注意：

Flink 不对窗口中元素的顺序做任何保证，即使 evictor 从窗口缓存的开头移除一个元素，这个元素也不一定是最先或者最后到达窗口的；

默认情况下，所有内置的 evictor 逻辑都在调用窗口函数前执行；

4、代码示例

package com.xu.flink.datastream.day09;import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.functions.windowing.delta.DeltaFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.evictors.DeltaEvictor;
import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;import java.time.Duration;/*** Evictor 可以在 trigger 触发后、调用窗口函数之前或之后从窗口中删除元素* evictBefore() 包含在调用窗口函数前的逻辑，在调用窗口函数之前被移除的元素不会被窗口函数计算* evictAfter() 包含在窗口函数调用之后的逻辑* <p>* -默认情况下，所有内置的 evictor 逻辑都在调用窗口函数前执行-* CountEvictor: 仅记录用户指定数量的元素，一旦窗口中的元素超过这个数量，多余的元素会从窗口缓存的开头移除。* DeltaEvictor: 接收 DeltaFunction 和 threshold 参数，计算最后一个元素与窗口缓存中所有元素的差值，并移除差值大于或等于 threshold 的元素。* TimeEvictor: 接收 interval 参数，以毫秒表示，它会找到窗口中元素的最大 timestamp `max_ts`，并移除比 `max_ts - interval` 小的所有元素。* <p>* 注意：Flink 不对窗口中元素的顺序做任何保证，即使 evictor 从窗口缓存的开头移除一个元素，这个元素也不一定是最先或者最后到达窗口的。*/
public class _12_WindowDefaultEvictors {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> input = env.socketTextStream("localhost", 8888);// 测试时限制了分区数，生产中需要设置空闲数据源env.setParallelism(2);// 事件时间需要设置水位线策略和时间戳SingleOutputStreamOperator<Tuple2<String, Long>> map = input.map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> map(String input) throws Exception {String[] fields = input.split(",");return new Tuple2<>(fields[0], Long.parseLong(fields[1]));}});SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = map.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(0)).withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {@Overridepublic long extractTimestamp(Tuple2<String, Long> input, long l) {return input.f1;}}));//1、CountEvictor: 仅记录用户指定数量的元素，一旦窗口中的元素超过这个数量，多余的元素会从窗口缓存的开头移除(默认)。//1.1 doEvictAfter = false//a,1718157600000//b,1718157600000//c,1718157600000////a,1718157602000//b,1718157602000//c,1718157602000////a,1718157604000//b,1718157604000//c,1718157604000////a,1718157605001//b,1718157605001////Window 的开始和结束时间=>1718157600000-1718157605000//2> a//Window 的开始和结束时间=>1718157600000-1718157605000//1> b//Window 的开始和结束时间=>1718157600000-1718157605000//1> c////c,1718157605001////1.2 doEvictAfter = true//a,1718157600000//b,1718157600000//c,1718157600000////a,1718157602000//b,1718157602000//c,1718157602000////a,1718157604000//b,1718157604000//c,1718157604000////a,1718157605001//b,1718157605001////Window 的开始和结束时间=>1718157600000-1718157605000//2> a//2> a//2> a//Window 的开始和结束时间=>1718157600000-1718157605000//1> b//1> b//1> b//Window 的开始和结束时间=>1718157600000-1718157605000//1> c//1> c//1> c////c,1718157605001////2、DeltaEvictor：接收 DeltaFunction 和 threshold 参数，计算最后一个元素与窗口缓存中所有元素的差值，并移除差值大于或等于 threshold 的元素。//a,1718157600000//b,1718157600000//c,1718157600000////a,1718157602000//b,1718157602000//c,1718157602000////a,1718157604000//b,1718157604000//c,1718157604000////a,1718157605001-最后的元素//b,1718157605001-最后的元素////first=>(a,1718157600000),last=>(a,1718157604000)//first=>(a,1718157602000),last=>(a,1718157604000)//first=>(a,1718157604000),last=>(a,1718157604000)//first=>(b,1718157600000),last=>(b,1718157604000)//first=>(b,1718157602000),last=>(b,1718157604000)//first=>(b,1718157604000),last=>(b,1718157604000)//first=>(c,1718157600000),last=>(c,1718157604000)//first=>(c,1718157602000),last=>(c,1718157604000)//first=>(c,1718157604000),last=>(c,1718157604000)////Window 的开始和结束时间=>1718157600000-1718157605000//Window 的开始和结束时间=>1718157600000-1718157605000//Window 的开始和结束时间=>1718157600000-1718157605000//2> a//2> a//2> a//c,1718157605001-最后的元素////3、TimeEvictor：接收 interval 参数，以毫秒表示，它会找到窗口中元素的最大 timestamp `max_ts`，并移除比 `max_ts - interval` 小的所有元素//a,1718157600000//b,1718157600000//c,1718157600000////a,1718157602000//b,1718157602000//c,1718157602000////a,1718157604000//b,1718157604000//c,1718157604000////a,1718157605001//b,1718157605001////Window 的开始和结束时间=>1718157600000-1718157605000//2> a//Window 的开始和结束时间=>1718157600000-1718157605000//1> b//Window 的开始和结束时间=>1718157600000-1718157605000//1> c////c,1718157605001watermarks.keyBy(e -> e.f0).window(TumblingEventTimeWindows.of(Duration.ofSeconds(5))).trigger(EventTimeTrigger.create()).evictor(TimeEvictor.of(Duration.ofSeconds(1), false))
//                .evictor(CountEvictor.of(1, true))
//                .evictor(DeltaEvictor.of(1.0, new DeltaFunction<Tuple2<String, Long>>() {
//                    @Override
//                    public double getDelta(Tuple2<String, Long> first, Tuple2<String, Long> last) {
//                        System.out.println("first=>"+first+",last=>"+last);
//                        if(first.f0.startsWith("a") && last.f0.startsWith("a")){
//                            return 0.0;
//                        }
//                        return 2.0;
//                    }
//                }, false)).apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {@Overridepublic void apply(String s, TimeWindow timeWindow, Iterable<Tuple2<String, Long>> iterable, Collector<String> collector) throws Exception {System.out.println("Window 的开始和结束时间=>" + timeWindow.getStart() + "-" + timeWindow.getEnd());for (Tuple2<String, Long> tuple2 : iterable) {collector.collect(tuple2.f0);}}}).print();env.execute();}
}

这篇关于44、Flink 的默认窗口剔除器 evictor 代码示例的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1060179。

相关文章

Python函数作用域示例详解

Python函数作用域示例详解

《Python函数作用域示例详解》本文介绍了Python中的LEGB作用域规则,详细解析了变量查找的四个层级,通过具体代码示例,展示了各层级的变量访问规则和特性,对python函数作用域相关知识感兴趣... 目录一、LEGB 规则二、作用域实例2.1 局部作用域（Local）2.2 闭包作用域（Enclos
阅读更多...
C++20管道运算符的实现示例

C++20管道运算符的实现示例

《C++20管道运算符的实现示例》本文简要介绍C++20管道运算符的使用与实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录标准库的管道运算符使用自己实现类似的管道运算符我们不打算介绍太多，因为它实际属于c++20最为重要的
阅读更多...
Java中调用数据库存储过程的示例代码

Java中调用数据库存储过程的示例代码

《Java中调用数据库存储过程的示例代码》本文介绍Java通过JDBC调用数据库存储过程的方法,涵盖参数类型、执行步骤及数据库差异,需注意异常处理与资源管理,以优化性能并实现复杂业务逻辑,感兴趣的朋友... 目录一、存储过程概述二、Java调用存储过程的基本javascript步骤三、Java调用存储过程示
阅读更多...
Visual Studio 2022 编译C++20代码的图文步骤

Visual Studio 2022 编译C++20代码的图文步骤

《VisualStudio2022编译C++20代码的图文步骤》在VisualStudio中启用C++20import功能,需设置语言标准为ISOC++20,开启扫描源查找模块依赖及实验性标... 默认创建Visual Studio桌面控制台项目代码包含C++20的import方法。右键项目的属性：
阅读更多...
ModelMapper基本使用和常见场景示例详解

ModelMapper基本使用和常见场景示例详解

《ModelMapper基本使用和常见场景示例详解》ModelMapper是Java对象映射库,支持自动映射、自定义规则、集合转换及高级配置（如匹配策略、转换器）,可集成SpringBoot,减少样板... 目录1. 添加依赖2. 基本用法示例：简单对象映射3. 自定义映射规则4. 集合映射5. 高级配置匹
阅读更多...
MySQL数据库的内嵌函数和联合查询实例代码

MySQL数据库的内嵌函数和联合查询实例代码

《MySQL数据库的内嵌函数和联合查询实例代码》联合查询是一种将多个查询结果组合在一起的方法,通常使用UNION、UNIONALL、INTERSECT和EXCEPT关键字,下面：本文主要介绍MyS... 目录一.数据库的内嵌函数1.1聚合函数COUNT([DISTINCT] expr)SUM([DISTIN
阅读更多...
C++11作用域枚举(Scoped Enums)的实现示例

C++11作用域枚举(Scoped Enums)的实现示例

《C++11作用域枚举(ScopedEnums)的实现示例》枚举类型是一种非常实用的工具,C++11标准引入了作用域枚举,也称为强类型枚举,本文主要介绍了C++11作用域枚举(ScopedEnums... 目录一、引言二、传统枚举类型的局限性2.1 命名空间污染2.2 整型提升问题2.3 类型转换问题三、C
阅读更多...
Java实现自定义table宽高的示例代码

Java实现自定义table宽高的示例代码

《Java实现自定义table宽高的示例代码》在桌面应用、管理系统乃至报表工具中,表格（JTable）作为最常用的数据展示组件,不仅承载对数据的增删改查,还需要配合布局与视觉需求,而JavaSwing... 目录一、项目背景详细介绍二、项目需求详细介绍三、相关技术详细介绍四、实现思路详细介绍五、完整实现代码
阅读更多...
Go语言代码格式化的技巧分享

Go语言代码格式化的技巧分享

《Go语言代码格式化的技巧分享》在Go语言的开发过程中,代码格式化是一个看似细微却至关重要的环节,良好的代码格式化不仅能提升代码的可读性,还能促进团队协作,减少因代码风格差异引发的问题,Go在代码格式... 目录一、Go 语言代码格式化的重要性二、Go 语言代码格式化工具：gofmt 与 go fmt（一）
阅读更多...
C++ 检测文件大小和文件传输的方法示例详解

C++ 检测文件大小和文件传输的方法示例详解

《C++检测文件大小和文件传输的方法示例详解》文章介绍了在C/C++中获取文件大小的三种方法,推荐使用stat()函数,并详细说明了如何设计一次性发送压缩包的结构体及传输流程,包含CRC校验和自动解... 目录检测文件的大小✅ 方法一：使用 stat() 函数（推荐）✅ 用法示例：✅ 方法二：使用 fsee
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2