ProcessWindowFunction 结合自定义触发器的陷阱
背景:
flink中常见的需求如下:统计某个页面一天内的点击率,每10秒输出一次,我们如果采用ProcessWindowFunction 结合自定义触发器如何实现呢?如果这样实现问题是什么呢?
ProcessWindowFunction 结合自定义触发器实现统计点击率
关键代码:


完整代码参见:
package wikiedits.func;import java.text.SimpleDateFormat;
import java.util.Date;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.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;import wikiedits.func.model.KeyCount;public class ProcessWindowFunctionAndTiggerDemo {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 使用处理时间env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);env.setStateBackend(new FsStateBackend("file:///D:/tmp/flink/checkpoint/windowtrigger"));// 并行度为1env.setParallelism(1);// 设置数据源,一共三个元素DataStream<Tuple2<String, Integer>> dataStream = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {@Overridepublic void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {int xxxNum = 0;int yyyNum = 0;for (int i = 1; i < Integer.MAX_VALUE; i++) {// 只有XXX和YYY两种nameString name = (0 == i % 2) ? "XXX" : "YYY";// 更新aaa和bbb元素的总数if (0 == i % 2) {xxxNum++;} else {yyyNum++;}// 使用当前时间作为时间戳long timeStamp = System.currentTimeMillis();// 将数据和时间戳打印出来,用来验证数据if(xxxNum % 2000==0){System.out.println(String.format("source,%s, %s, XXX total : %d, YYY total : %d\n", name,time(timeStamp), xxxNum, yyyNum));}// 发射一个元素,并且戴上了时间戳ctx.collectWithTimestamp(new Tuple2<String, Integer>(name, 1), timeStamp);// 每发射一次就延时1秒Thread.sleep(1);}}@Overridepublic void cancel() {}});// 将数据用5秒的滚动窗口做划分,再用ProcessWindowFunctionSingleOutputStreamOperator<String> mainDataStream = dataStream// 以Tuple2的f0字段作为key,本例中实际上key只有aaa和bbb两种.keyBy(value -> value.f0)// 5秒一次的滚动窗口.timeWindow(Time.minutes(5))// 10s触发一次计算,更新统计结果.trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))// 统计每个key当前窗口内的元素数量,然后把key、数量、窗口起止时间整理成字符串发送给下游算子.process(new ProcessWindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {// 自定义状态private ValueState<KeyCount> state;@Overridepublic void open(Configuration parameters) throws Exception {// 初始化状态,name是myStatestate = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", KeyCount.class));}public void clear(Context context) {ValueState<KeyCount> contextWindowValueState = context.windowState().getState(new ValueStateDescriptor<>("myWindowState", KeyCount.class));contextWindowValueState.clear();}@Overridepublic void process(String s, Context context, Iterable<Tuple2<String, Integer>> iterable,Collector<String> collector) throws Exception {// 从backend取得当前单词的myState状态KeyCount current = state.value();// 如果myState还从未没有赋值过,就在此初始化if (current == null) {current = new KeyCount();current.key = s;current.count = 0;}int count = 0;// iterable可以访问该key当前窗口内的所有数据,// 这里简单处理,只统计了元素数量for (Tuple2<String, Integer> tuple2 : iterable) {count++;}// 更新当前key的元素总数current.count += count;// 更新状态到backendstate.update(current);ValueState<KeyCount> contextWindowValueState = context.windowState().getState(new ValueStateDescriptor<>("myWindowState", KeyCount.class));KeyCount windowValue = contextWindowValueState.value();if (windowValue == null) {windowValue = new KeyCount();windowValue.key = s;windowValue.count = 0;}windowValue.count += count;contextWindowValueState.update(windowValue);// 将当前key及其窗口的元素数量,还有窗口的起止时间整理成字符串String value = String.format("window, %s, %s - %s, %d, windowStateCount :%d, total : %d",// 当前keys,// 当前窗口的起始时间time(context.window().getStart()),// 当前窗口的结束时间time(context.window().getEnd()),// 当前key在当前窗口内元素总数count,// 当前key所在窗口的总数contextWindowValueState.value().count,// 当前key出现的总数current.count);// 发射到下游算子collector.collect(value);}});// 打印结果,通过分析打印信息,检查ProcessWindowFunction中可以处理所有key的整个窗口的数据mainDataStream.print();env.execute("processfunction demo : processwindowfunction");}public static String time(long timeStamp) {return new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date(timeStamp));}}
这里采用ProcessWindowFunction 结合ContinuousProcessingTimeTrigger的方式确实可以实现统计至今为止某个页面点击率的目的,不过这其中需要注意点的点是:
每隔10s触发public void process(String s, Context context, Iterable<Tuple2<String, Integer>> iterable, Collector<String> collector)方法时,iterable对象是包含了一天的窗口内收到的所有消息,也就是当前触发时iterable集合是前10s触发时iterable集合的超集,包含前10s触发时的所有的消息集合。
到这里所引起的问题也自然而然的出来了:对于ProcessWindowFunction 实现而言,flink内部是通过ListState的形式保存窗口内收到的所有消息的,注意这里flink内部会使用ListState保存每一条分配到以天为单位的窗口内的消息,这会导致状态膨胀,想一下,一天内所有的消息都会当成状态保存起来,这对于状态后端的压力是有多大!这些保存在ListState中的消息只有在窗口结束后才会清理:具体参见WindowOperator.clearAllState,那有解决方案吗?使用Agg/Reduce处理函数替ProcessWindowFunction作为处理函数可以实现吗?请看下一篇文章
参考文章:
https://www.cnblogs.com/Springmoon-venn/p/13667023.html
相关文章:
ProcessWindowFunction 结合自定义触发器的陷阱
背景: flink中常见的需求如下:统计某个页面一天内的点击率,每10秒输出一次,我们如果采用ProcessWindowFunction 结合自定义触发器如何实现呢?如果这样实现问题是什么呢? ProcessWindowFunction 结合自定义触发器实现…...
什么是jvm
一、初识JVM(虚拟机) JVM是Java Virtual Machine(Java虚拟机)的缩写,JVM是一种用于计算设备的规范,它是一个虚构出来的计算机,是通过在实际的计算机上仿真模拟各种计算机功能来实现的。 引入Jav…...
kettle通过java步骤获取汉字首拼
kettle通过java步骤获取汉字首拼 用途描述 一组数据,需要获取汉字首拼后,输出; 实现效果 添加jar包 pinyin4j-2.5.0.jar 自定义常量数据 Java代码 完整代码: import net.sourceforge.pinyin4j.PinyinHelper; import net.sou…...
Conformer: Local Features Coupling Global Representationsfor Visual Recognition
论文链接:https://arxiv.org/abs/2105.03889 代码链接:https://github.com/pengzhiliang/Conformer 参考博文:Conformer论文以及代码解析(上)_conformer代码_从现在开始壹并超的博客-CSDN博客 摘要 在卷积神经网络…...
java8-Stream流常用API
什么是 Stream Stream(流)是 Java 8 引入的一个新的抽象概念,它代表着一种处理数据的序列。简单来说,Stream 是一系列元素的集合,这些元素可以是集合、数组、I/O 资源或者其他数据源。 Stream API 提供了丰富的操作方…...
React 任务调度
React 任务池 不同的fiber任务有不同的优先级,为了用户体验,React需要先处理优先级高的任务。 为了存储这些任务,React中有两个任务池: // Tasks are stored on a min heap var taskQueue []; // 存储立即要执行的任务 var tim…...
小白开始学习C++
第一节:控制台输出hello word! #include<iostream> //引入库文件 int main() { //控制台输出 hello word! 之后回车 std::cout << "hello word!\n"; #include<iostream> //引入库文件int main() {//控制…...
SpringMVC入门的注解、参数传递、返回值和页面跳转---超详细教学
前言: 欢迎阅读Spring MVC入门必读!在这篇文章中,我们将探索这个令人兴奋的框架,它为您提供了一种高效、灵活且易于维护的方式来构建Web应用程序。通过使用Spring MVC,您将享受到以下好处:简洁的代码、强大…...
【复习socket】每天40min,我们一起用70天稳扎稳打学完《JavaEE初阶》——28/70 第二十八天
专注 效率 记忆 预习 笔记 复习 做题 欢迎观看我的博客,如有问题交流,欢迎评论区留言,一定尽快回复!(大家可以去看我的专栏,是所有文章的目录) 文章字体风格: 红色文字表示:重难点★✔ 蓝色文字表示:思路以及想法★✔ 如果大家觉得有帮助的话,感谢大家帮忙 点…...
vue2踩坑之项目:生成二维码使用vue-print-nb打印二维码
1. vue2安装 npm install vue-print-nb --save vue3安装 npm install vue3-print-nb --save 2. //vue2 引入方式 全局 main.js import Print from vue-print-nb Vue.use(Print) ------------------------------------------------------------------------------------ //vue2 …...
【iVX】十五分钟制作一款小游戏,iVX真有怎么神?
个人主页:【😊个人主页】 新人博主,喜欢就关注一下呗~ 文章目录 前言iVX介绍初上手布置背景制作可移动物体总结(完善步骤) 前言 在上篇文章中,我向大家介绍了一种打破常规的编程方式——iVX,可…...
SpringMVC常用注解、参数传递、返回值
目录 前言 一、常用注解 二、参数传递 编辑 1. 基础类型String类型 2. 复杂类型 3. RequestParam 4. PathVariable 5.RequestBody 6. RequestHeader 三、方法返回值 一:void 二:String 三:Stringmodel 四:ModelAndVi…...
新公司第一次上架新APP需要提前准备哪些材料?
目录 前言一、需要上架的应用市场二、需要准备的资料总结 前言 前不久,使用一家新公司刚刚上架了一款新的APP项目。特此记录一下,现在第一次上架一款APP需要提前准备的各项材料。 一、需要上架的应用市场 现在,上架一款新的APP主流的应用市…...
『C语言进阶』指针进阶(一)
🔥博客主页: 小羊失眠啦 🔖系列专栏: C语言 🌥️每日语录:无论你怎么选,都难免会有遗憾。 ❤️感谢大家点赞👍收藏⭐评论✍️ 前言 在C语言初阶中,我们对指针有了一定的…...
2605. 从两个数字数组里生成最小数字(Java)
给你两个只包含 1 到 9 之间数字的数组 nums1 和 nums2 ,每个数组中的元素 互不相同 ,请你返回 最小 的数字,两个数组都 至少 包含这个数字的某个数位。 示例 1: 输入:nums1 [4,1,3], nums2 [5,7] 输出:1…...
深度解析 PostgreSQL Protocol v3.0(一)
引言 PostgreSQL 使用基于消息的协议在前端(也可以称为客户端)和后端(也可以称为服务器)之间进行通信。该协议通过 TCP/IP 和 Unix 域套接字支持。 《深度解析 PostgreSQL Protocol v3.0》系列技术贴,将带大家深度了…...
Mysql中having语句与where语句的用法与区别
分析&回答 我们在写sql语句的时候,经常会使用where语句,很少会用到having,其实在mysql中having子句也是设定条件的语句与where有相似之处但也有区别。having子句在查询过程中慢于聚合语句(sum,min,max,avg,count)。而where子句在查询过程中则快于聚合语句(sum,min,max,avg…...
基于qt软件的网上聊天室软件
1.服务器: 1).功能: 用于创建一个客户端,通过文本编辑器来获得端口号,根据获得的端口号创建服务器,等待客户端连接 创建成功会提示服务器创建成功 在收到客户端发送的信息时,把这条信息发送给其他所有客户端,实现群…...
本是同根生-双数据库集群keepalived virtual_route_id冲突导致连接故障
项目场景: 一企业近期陆续开始升级办公与大数据系统,新的承包商。原有的数据库是某国内大品牌A,现在新的功能准备陆续迁移到大品牌B上。系统部署后,A依旧承担比较轻松的财务、仓库管理,B承担实时的线上业务。项目验收…...
『力扣每日一题06』字符串中的第一个唯一字符
今天是学习新知识的一天,String 类中有太多细枝末节,需要我去学习跟掌握了。 话不多说,今天给大家带来一道字符串的题目~ 一、题目 给定一个字符串 s ,找到 它的第一个不重复的字符,并返回它的索引 。如果不存在&…...
Golang dig框架与GraphQL的完美结合
将 Go 的 Dig 依赖注入框架与 GraphQL 结合使用,可以显著提升应用程序的可维护性、可测试性以及灵活性。 Dig 是一个强大的依赖注入容器,能够帮助开发者更好地管理复杂的依赖关系,而 GraphQL 则是一种用于 API 的查询语言,能够提…...
2025 后端自学UNIAPP【项目实战:旅游项目】6、我的收藏页面
代码框架视图 1、先添加一个获取收藏景点的列表请求 【在文件my_api.js文件中添加】 // 引入公共的请求封装 import http from ./my_http.js// 登录接口(适配服务端返回 Token) export const login async (code, avatar) > {const res await http…...
c#开发AI模型对话
AI模型 前面已经介绍了一般AI模型本地部署,直接调用现成的模型数据。这里主要讲述讲接口集成到我们自己的程序中使用方式。 微软提供了ML.NET来开发和使用AI模型,但是目前国内可能使用不多,至少实践例子很少看见。开发训练模型就不介绍了&am…...
2023赣州旅游投资集团
单选题 1.“不登高山,不知天之高也;不临深溪,不知地之厚也。”这句话说明_____。 A、人的意识具有创造性 B、人的认识是独立于实践之外的 C、实践在认识过程中具有决定作用 D、人的一切知识都是从直接经验中获得的 参考答案: C 本题解…...
Typeerror: cannot read properties of undefined (reading ‘XXX‘)
最近需要在离线机器上运行软件,所以得把软件用docker打包起来,大部分功能都没问题,出了一个奇怪的事情。同样的代码,在本机上用vscode可以运行起来,但是打包之后在docker里出现了问题。使用的是dialog组件,…...
React---day11
14.4 react-redux第三方库 提供connect、thunk之类的函数 以获取一个banner数据为例子 store: 我们在使用异步的时候理应是要使用中间件的,但是configureStore 已经自动集成了 redux-thunk,注意action里面要返回函数 import { configureS…...
让回归模型不再被异常值“带跑偏“,MSE和Cauchy损失函数在噪声数据环境下的实战对比
在机器学习的回归分析中,损失函数的选择对模型性能具有决定性影响。均方误差(MSE)作为经典的损失函数,在处理干净数据时表现优异,但在面对包含异常值的噪声数据时,其对大误差的二次惩罚机制往往导致模型参数…...
深入浅出深度学习基础:从感知机到全连接神经网络的核心原理与应用
文章目录 前言一、感知机 (Perceptron)1.1 基础介绍1.1.1 感知机是什么?1.1.2 感知机的工作原理 1.2 感知机的简单应用:基本逻辑门1.2.1 逻辑与 (Logic AND)1.2.2 逻辑或 (Logic OR)1.2.3 逻辑与非 (Logic NAND) 1.3 感知机的实现1.3.1 简单实现 (基于阈…...
Linux nano命令的基本使用
参考资料 GNU nanoを使いこなすnano基础 目录 一. 简介二. 文件打开2.1 普通方式打开文件2.2 只读方式打开文件 三. 文件查看3.1 打开文件时,显示行号3.2 翻页查看 四. 文件编辑4.1 Ctrl K 复制 和 Ctrl U 粘贴4.2 Alt/Esc U 撤回 五. 文件保存与退出5.1 Ctrl …...
android RelativeLayout布局
<?xml version"1.0" encoding"utf-8"?> <RelativeLayout xmlns:android"http://schemas.android.com/apk/res/android"android:layout_width"match_parent"android:layout_height"match_parent"android:gravity&…...
