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

2024-06-14 10:44
文章标签 示例 44 flink 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。

相关文章

uniapp接入微信小程序原生代码配置方案（优化版）

uniapp接入微信小程序原生代码配置方案（优化版）

uniapp项目需要把微信小程序原生语法的功能代码嵌套过来，无需把原生代码转换为uniapp，可以配置拷贝的方式集成过来 1、拷贝代码包到src目录 2、vue.config.js中配置原生代码包直接拷贝到编译目录中 3、pages.json中配置分包目录，原生入口组件的路径 4、manifest.json中配置分包，使用原生组件 5、需要把原生代码包里的页面修改成组件的方
阅读更多...
公共筛选组件（二次封装antd）支持代码提示

公共筛选组件（二次封装antd）支持代码提示

如果项目是基于antd组件库为基础搭建，可使用此公共筛选组件 使用到的库 npm i antdnpm i lodash-esnpm i @types/lodash-es -D /components/CommonSearch index.tsx import React from 'react';import { Button, Card, Form } from 'antd'
阅读更多...
17.用300行代码手写初体验Spring V1.0版本

17.用300行代码手写初体验Spring V1.0版本

1.1.课程目标 1、了解看源码最有效的方式，先猜测后验证，不要一开始就去调试代码。 2、浓缩就是精华，用 300行最简洁的代码 提炼Spring的基本设计思想。 3、掌握Spring框架的基本脉络。 1.2.内容定位 1、 具有1年以上的SpringMVC使用经验。 2、 希望深入了解Spring源码的人群，对 Spring有一个整体的宏观感受。 3、 全程手写实现SpringM
阅读更多...
代码随想录算法训练营：12/60

代码随想录算法训练营：12/60

非科班学习算法day12 | LeetCode150:逆波兰表达式 ，Leetcode239: 滑动窗口最大值  目录 介绍 一、基础概念补充： 1.c++字符串转为数字 1. std::stoi, std::stol, std::stoll, std::stoul, std::stoull（最常用） 2. std::stringstream 3. std::atoi, std
阅读更多...
记录AS混淆代码模板

记录AS混淆代码模板

开启混淆得先在build.gradle文件中把 minifyEnabled false改成true,以及shrinkResources true//去除无用的resource文件 这些是写在proguard-rules.pro文件内的 指定代码的压缩级别 -optimizationpasses 5 包明不混合大小写 -dontusemixedcaseclassnames 不去忽略非公共
阅读更多...
麻了！一觉醒来，代码全挂了。。

麻了！一觉醒来，代码全挂了。。

作为⼀名程序员，相信大家平时都有代码托管的需求。 相信有不少同学或者团队都习惯把自己的代码托管到GitHub平台上。 但是GitHub大家知道，经常在访问速度这方面并不是很快，有时候因为网络问题甚至根本连网站都打不开了，所以导致使用体验并不友好。 经常一觉醒来，居然发现我竟然看不到我自己上传的代码了。。 那在国内，除了GitHub，另外还有一个比较常用的Gitee平台也可以用于
阅读更多...
众所周知，配置即代码≠基础设置即代码

众所周知，配置即代码≠基础设置即代码

​前段时间翻到几条留言，问： “配置即代码和基础设施即代码一样吗？” “配置即代码是什么？怎么都是基础设施即代码？” 我们都是知道，DevOp的快速发展，让服务器管理与配置的时间大大减少，配置即代码和基础设施即代码作为DevOps的重要实践，在其中起到了关键性作用。 不少人将二者看作是一件事，配置即大代码是关于管理特定的应用程序配置设置本身，而基础设施即代码更关注的是部署支持应用程序环境所需的
阅读更多...
53、Flink Interval Join 代码示例

53、Flink Interval Join 代码示例

1、概述 interval Join 默认会根据 keyBy 的条件进行 Join 此时为 Inner Join； interval Join 算子的水位线会取两条流中水位线的最小值； interval Join 迟到数据的判定是以 interval Join 算子的水位线为基准； interval Join 可以分别输出两条流中迟到的数据-[sideOutputLeftLateData,
阅读更多...
【Java算法】滑动窗口 下

【Java算法】滑动窗口 下

​ ​    🔥个人主页： 中草药 🔥专栏：【算法工作坊】算法实战揭秘 🦌一.水果成篮 题目链接：904.水果成篮 ​ 算法原理 算法原理是使用“滑动窗口”（Sliding Window）策略，结合哈希表（Map）来高效地统计窗口内不同水果的种类数量。以下是详细分析： 初始化：创建一个空的哈希表 map 用来存储每种水果的数量，初始化左右指针 left
阅读更多...
TableView 当前选中的行号。 默认会使哪一行选中 加入导航条后contentInset向下偏移的64

TableView 当前选中的行号。 默认会使哪一行选中 加入导航条后contentInset向下偏移的64

1.得到当前选中的行号     NSLog(@"%ld %s",  [self.tableView indexPathForSelectedRow].row,__func__); 2.默认选中表格的那一行     [self.tableView selectRowAtIndexPath:[NSIndexPath indexPathForRow:0 inSection:0 ]
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2