一文弄明白KeyedProcessFunction函数
引言
KeyedProcessFunction是Flink用于处理KeyedStream的数据集合,它比ProcessFunction拥有更多特性,例如状态处理和定时器功能等。接下来就一起来了解下这个函数吧
正文
了解一个函数怎么用最权威的地方就是 官方文档 以及注解,KeyedProcessFunction的注解如下
/*** A keyed function that processes elements of a stream.** <p>For every element in the input stream {@link #processElement(Object, Context, Collector)} is* invoked. This can produce zero or more elements as output. Implementations can also query the* time and set timers through the provided {@link Context}. For firing timers {@link #onTimer(long,* OnTimerContext, Collector)} will be invoked. This can again produce zero or more elements as* output and register further timers.** <p><b>NOTE:</b> Access to keyed state and timers (which are also scoped to a key) is only* available if the {@code KeyedProcessFunction} is applied on a {@code KeyedStream}.** <p><b>NOTE:</b> A {@code KeyedProcessFunction} is always a {@link* org.apache.flink.api.common.functions.RichFunction}. Therefore, access to the {@link* org.apache.flink.api.common.functions.RuntimeContext} is always available and setup and teardown* methods can be implemented. See {@link* org.apache.flink.api.common.functions.RichFunction#open(org.apache.flink.configuration.Configuration)}* and {@link org.apache.flink.api.common.functions.RichFunction#close()}.*/
上面简单来说就是以下四点
- Flink中输入流中的每一条数据都会触发KeyedProcessFunction类的processElement方法调用
- 通过这个方法的Context参数可以设置定时器,在开启定时器后会程序会定时调用onTimer方法
- 由于KeyedProcessFunction实现了RichFunction接口,因此是可以通过RuntimeContext上下文对象管理状态state的开启和释放
- 需要注意的是,只有在KeyedStream里才能够访问state和定时器,通俗点来说就是这个函数要用在keyBy这个函数的后面
processElement方法解析
- Flink会调用processElement方法处理输入流中的每一条数据
- KeyedProcessFunction.Context参数可以用来读取以及更新内部状态state
- 这个KeyedProcessFunction跟其他function一样通过参数中的Collector对象以回写的方式返回数据
onTimer方法解析:在启用TimerService服务时会定时触发此方法,一般会在processElement方法中开启TimerService服务
以上就是这个函数的基本知识,接下来就通过实战来熟悉下它的使用
实战简介
本次实战的目标是学习KeyedProcessFunction,内容如下:
- 监听本机7777端口读取字符串
- 将每个字符串用空格分隔,转成Tuple2实例,f0是分隔后的单词,f1等于1
- 将Tuple2实例集合通过f0字段分区,得到KeyedStream
- KeyedSteam通过自定义KeyedProcessFunction处理
- 自定义KeyedProcessFunction的作用,是记录每个单词最新一次出现的时间,然后建一个十秒的定时器进行触发
使用代码例子
首先定义pojo类
public class CountWithTimestampNew {private String key;private long count;private long lastQuestTimestamp;public long getAndIncrementCount() {return ++count;}public String getKey() {return key;}public void setKey(String key) {this.key = key;}public long getCount() {return count;}public void setCount(long count) {this.count = count;}public long getLastQuestTimestamp() {return lastQuestTimestamp;}public void setLastQuestTimestamp(long lastQuestTimestamp) {this.lastQuestTimestamp = lastQuestTimestamp;}
}
接着实现KeyedProcessFunction类
public class CountWithTimeoutKeyProcessFunctionNew extends KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Long>> {private ValueState<CountWithTimestampNew> state;@Overridepublic void open(Configuration parameters) throws Exception {state = getRuntimeContext().getState(new ValueStateDescriptor<CountWithTimestampNew>("sherlock-state", CountWithTimestampNew.class));}// 实现数据处理逻辑的地方@Overridepublic void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception {Tuple currentKey = ctx.getCurrentKey();CountWithTimestampNew countWithTimestampNew = state.value();if (countWithTimestampNew == null) {countWithTimestampNew = new CountWithTimestampNew();countWithTimestampNew.setKey(value.f0);}countWithTimestampNew.getAndIncrementCount();//更新这个单词最后一次出现的时间countWithTimestampNew.setLastQuestTimestamp(ctx.timestamp());//单词之间不会互相覆盖吗?推测state对象是跟key绑定,针对每一个不同的key KeyedProcessFunction会创建其对应的state对象state.update(countWithTimestampNew);//给当前单词创建定时器,十秒后触发long timer = countWithTimestampNew.getLastQuestTimestamp()+10000;//尝试注释掉看看是否还会触发onTimer方法ctx.timerService().registerProcessingTimeTimer(timer);//打印所有信息,用于确保数据准确性System.out.println(String.format(" 触发processElement方法,当前的key是 %s, 这个单词累加次数是 %d, 上次请求的时间是:%s, timer的时间是: %s",currentKey.getField(0),countWithTimestampNew.getCount(),time(countWithTimestampNew.getLastQuestTimestamp()),time(timer)));}@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out) throws Exception {Tuple currentKey = ctx.getCurrentKey();CountWithTimestampNew countWithTimestampNew = state.value();//标记当前元素是否已经连续10s未出现boolean isTimeout = false;if (timestamp >= countWithTimestampNew.getLastQuestTimestamp()+10000 ) {//out.collect(new Tuple2<>(countWithTimestampNew.getKey(), countWithTimestampNew.getCount()));isTimeout = true;}//打印所有信息,用于确保数据准确性System.out.println(String.format(" 触发onTimer方法,当前的key是 %s, 这个单词累加次数是 %d, 上次请求的时间是:%s, timer的时间是: %s, 当前单词是否已超过10秒没有再请求: %s",currentKey.getField(0),countWithTimestampNew.getCount(),time(countWithTimestampNew.getLastQuestTimestamp()),time(timestamp),String.valueOf(isTimeout)));}public static String time(long timeStamp) {return new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date(timeStamp));}
}
最后是启动类
public class KeyedProcessFunctionDemo2 {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 并行度1env.setParallelism(1);// 处理时间env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);// 监听本地9999端口,读取字符串DataStream<String> socketDataStream = env.socketTextStream("localhost", 7777);// 所有输入的单词,如果超过10秒没有再次出现,都可以通过CountWithTimeoutFunction得到DataStream<Tuple2<String, Long>> timeOutWord = socketDataStream// 对收到的字符串用空格做分割,得到多个单词.flatMap(new SplitterFlatMapFunction())// 设置时间戳分配器,用当前时间作为时间戳.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {@Overridepublic long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {// 使用当前系统时间作为时间戳return System.currentTimeMillis();}@Overridepublic Watermark getCurrentWatermark() {// 本例不需要watermark,返回nullreturn null;}})// 将单词作为key分区.keyBy(0)// 按单词分区后的数据,交给自定义KeyedProcessFunction处理.process(new CountWithTimeoutKeyProcessFunctionNew());// 所有输入的单词,如果超过10秒没有再次出现,就在此打印出来timeOutWord.print();env.execute("ProcessFunction demo : KeyedProcessFunction");}
}
演示
在启动服务前,先通过linux指令监听端口 nc -lk 7777
-
启动Flink服务后,往7777端口里面发送数据

-
通过IDEA的终端可以看到有日志输出,可以看到在发送消息的时候第一条日志立马打印出来并在10秒后输出第二条日志

-
那么咱们尝试连续发送两条Hello呢,可以看到累加器会持续累加,并且会触发两次onTimer方法,也就是每一条消息都会触发一次。由于连续发送两条,因此可以看得到第三行日志的末尾是false,说明收到第一条后的10秒内又有相同的消息进来。第二条是ture说明在收到第二条消息后的10秒内没有消息进来

-
再输入点其他的试试

-
通过输出可以看到这些单词的计数器又从0开始,说明每一个Key都对应一个状态

思考题
- open方法会在哪里进行调用,KeyedProcessFunction整个类的完整调用逻辑是怎么样的
- registerProcessingTimeTimer和registerEventTimeTimer的差异是什么
参考资料
- https://blog.csdn.net/boling_cavalry/article/details/106299167
- https://blog.csdn.net/lujisen/article/details/105510532
- https://blog.csdn.net/qq_31866793/article/details/102831731
相关文章:
一文弄明白KeyedProcessFunction函数
引言 KeyedProcessFunction是Flink用于处理KeyedStream的数据集合,它比ProcessFunction拥有更多特性,例如状态处理和定时器功能等。接下来就一起来了解下这个函数吧 正文 了解一个函数怎么用最权威的地方就是 官方文档 以及注解,KeyedProc…...
alibabacloud学习笔记06(小滴课堂)
讲Sentinel流量控制详细操作 基于并发线程进行限流配置实操 在浏览器打开快速刷新会报错 基于并发线程进行限流配置实操 讲解 微服务高可用利器Sentinel熔断降级规则 讲解服务调用常见的熔断状态和恢复 讲解服务调用熔断例子 我们写一个带异常的接口:...
Code Composer Studio (CCS) - Licensing Information
Code Composer Studio [CCS] - Licensing Information 1. Help -> Code Composer Studio Licensing Information2. Upgrade3. Specify a license fileReferences 1. Help -> Code Composer Studio Licensing Information 2. Upgrade 3. Specify a license file …...
uniapp引入微信小程序直播组件
方法1.小程序跳转视频号直播 微信小程序跳转到视频号 1.1微信开放平台注册 https://open.weixin.qq.com/ 2.2 方法2.使用小程序提供的直播组件 参考 微信小程序跳转视频号直播 小程序直播官方文档 https://developers.weixin.qq.com/miniprogram/dev/component/live-play…...
五个简单的C#编程案例
案例一:Hello, World! csharp using System; class Program { static void Main() { Console.WriteLine("Hello, World!"); } } 这个案例是最基础的C#程序,它打印出“Hello, World!”到控制台。每个C#程…...
Zlibrary低调官宣2024年最新网址,国内可直接访问,免费下载海量电子书籍
最近过节,文章也没怎么写,明天要上班了,今天写篇文章做个预热。 春节期间,“知识大航海”群里,有位群友分享了一个Zlibrary的最新地址,感谢这位群友妹妹的热心分享,这个地址国内可以直接访问。 …...
Android 开机启动
一、添加权限 <uses-permission android:name"android.permission.RECEIVE_BOOT_COMPLETED"/> 二、写一个广播接收器 public class BootReceiver extends BroadcastReceiver {Overridepublic void onReceive(Context context, Intent intent) {if(Intent.ACT…...
二叉树相关算法需了解汇总-基础算法操作
文章目录 144.二叉树的前序遍历145.二叉树的后序遍历94.二叉树的中序遍历102.二叉树的层序遍历107.二叉树的层次遍历倒序199.二叉树的右视图637.二叉树的层平均值429.N叉树的层序遍历515.在每个树行中找最大值116.填充每个节点的下一个右侧节点指针104.二叉树的最大深度111.二叉…...
万字干货-京东零售数据资产能力升级与实践
开篇 京东自营和商家自运营模式,以及伴随的多种运营视角、多种组合计算、多种销售属性等数据维度,相较于行业同等量级,数据处理的难度与复杂度都显著增加。如何从海量的数据模型与数据指标中提升检索数据的效率,降低数据存算的成…...
探索前端框架的世界:一场前端之旅
在网络世界中,网页开发领域的一颗明星是前端框架。这些框架为开发者提供了丰富的工具和技术,帮助他们构建出漂亮、高效的网页应用。现在,让我们随着小明的故事一起来探索一下吧。 小明的梦想 小明是一位年轻有为的前端开发者,他…...
class complex
class complex from C_OOP_base1_houjie complex.h #ifndef __COMPLEX__ // 防卫式声明 guard; 名称自定义 #define __COMPLEX__// 0. forward declarations class complex;complex& __doapl (complex* ths, const complex& r);// 1. class declarations class compl…...
数据库系统概论整理与总结
数据库系统概论 第一章:绪论 四个基本概念 四个概念 数据:Data 数据库:DataBase 数据库管理系统:DBMS 数据库系统:DBS 打个比喻,比如说菜鸟物流: Data:快递 DB:物流厂库 DBMS:对…...
打通新势力NAS权限壁垒,绿联私有云安装Portainer,实现更强大的Docker功能
打通新势力NAS权限壁垒,绿联私有云安装Portainer,实现更强大的Docker功能 对于国产新势力NAS来说,因为安全问题并没有完全开放SSH权限,所以还不能和传统NAS那样直接通过Docker run命令来部署容器,同时,对于…...
前端基础自学整理|DOM树
DOM,文档对象模型(Document Object Model),简单的说,DOM是一种理念,一种思想,一个与系统平台和编程语言无关的接口,一种方法, 使 Web开发人员可以访问HTML元素!不是具体方…...
RedisDesktopManager无法远程连接到Linux虚拟机中Redis的docker容器的一种解决方案
1.问题描述 除了RedisDesktopManager以外,使用java代码也无法连接到centos7虚拟机中的docker容器中的Redis ,按照网上其他博主的解决方案,在排除Linux防火墙问题,端口映射问题,redis.conf配置文件问题以后,…...
HarmonyOS 权限 介绍
权限说明 权限等级 根据权限对于不同等级应用有不同的开放范围,权限类型对应分为以下三种,等级依次提高。 normal权限 normal 权限允许应用访问超出默认规则外的普通系统资源。 这些系统资源的开放(包括数据和功能)对用户隐私以及…...
算法训练营day33(补),复习二叉树1
// 889. 根据前序和后序遍历构造二叉树 // 前序中左右 后序遍历左右中 func constructFromPrePost(preorder []int, postorder []int) *TreeNode { if len(preorder) 0 { return nil } root : &TreeNode{} root.Val preorder[0] //前序数组去掉root节点 preorder pre…...
k8s-权限管理
1. 身份认证 我们在目前的k8s集群环境里面,只能在master节点上执行kubectl的一些命令,在其他节点上执行就会报错 # 看一下是不是 [rootnode1 ~]# kubectl get nodes E0220 12:50:15.695133 6091 memcache.go:238] couldnt get current server API gro…...
四.QT5工具安装和环境变量的配置
1.以管理员身份运行安装包 2.登录qt账号,点击【next】 3.选中同意 4.选择安装目录,注意不能有中文和空格 5.勾选 64位 mingw。点击【next】,等待安装完成 6.配置环境变量...
为什么需要MDL锁
点击上方蓝字关注我 在数据库管理中,元数据(metadata)的保护至关重要,而MySQL中的"元数据锁"(MDL锁)就是它的守护者。 1. 什么是MDL锁MDL锁,全名Metadata Lock,是MySQL中…...
SpringBoot-17-MyBatis动态SQL标签之常用标签
文章目录 1 代码1.1 实体User.java1.2 接口UserMapper.java1.3 映射UserMapper.xml1.3.1 标签if1.3.2 标签if和where1.3.3 标签choose和when和otherwise1.4 UserController.java2 常用动态SQL标签2.1 标签set2.1.1 UserMapper.java2.1.2 UserMapper.xml2.1.3 UserController.ja…...
挑战杯推荐项目
“人工智能”创意赛 - 智能艺术创作助手:借助大模型技术,开发能根据用户输入的主题、风格等要求,生成绘画、音乐、文学作品等多种形式艺术创作灵感或初稿的应用,帮助艺术家和创意爱好者激发创意、提高创作效率。 - 个性化梦境…...
【WiFi帧结构】
文章目录 帧结构MAC头部管理帧 帧结构 Wi-Fi的帧分为三部分组成:MAC头部frame bodyFCS,其中MAC是固定格式的,frame body是可变长度。 MAC头部有frame control,duration,address1,address2,addre…...
2024年赣州旅游投资集团社会招聘笔试真
2024年赣州旅游投资集团社会招聘笔试真 题 ( 满 分 1 0 0 分 时 间 1 2 0 分 钟 ) 一、单选题(每题只有一个正确答案,答错、不答或多答均不得分) 1.纪要的特点不包括()。 A.概括重点 B.指导传达 C. 客观纪实 D.有言必录 【答案】: D 2.1864年,()预言了电磁波的存在,并指出…...
Auto-Coder使用GPT-4o完成:在用TabPFN这个模型构建一个预测未来3天涨跌的分类任务
通过akshare库,获取股票数据,并生成TabPFN这个模型 可以识别、处理的格式,写一个完整的预处理示例,并构建一个预测未来 3 天股价涨跌的分类任务 用TabPFN这个模型构建一个预测未来 3 天股价涨跌的分类任务,进行预测并输…...
AI编程--插件对比分析:CodeRider、GitHub Copilot及其他
AI编程插件对比分析:CodeRider、GitHub Copilot及其他 随着人工智能技术的快速发展,AI编程插件已成为提升开发者生产力的重要工具。CodeRider和GitHub Copilot作为市场上的领先者,分别以其独特的特性和生态系统吸引了大量开发者。本文将从功…...
CRMEB 框架中 PHP 上传扩展开发:涵盖本地上传及阿里云 OSS、腾讯云 COS、七牛云
目前已有本地上传、阿里云OSS上传、腾讯云COS上传、七牛云上传扩展 扩展入口文件 文件目录 crmeb\services\upload\Upload.php namespace crmeb\services\upload;use crmeb\basic\BaseManager; use think\facade\Config;/*** Class Upload* package crmeb\services\upload* …...
Unit 1 深度强化学习简介
Deep RL Course ——Unit 1 Introduction 从理论和实践层面深入学习深度强化学习。学会使用知名的深度强化学习库,例如 Stable Baselines3、RL Baselines3 Zoo、Sample Factory 和 CleanRL。在独特的环境中训练智能体,比如 SnowballFight、Huggy the Do…...
Linux --进程控制
本文从以下五个方面来初步认识进程控制: 目录 进程创建 进程终止 进程等待 进程替换 模拟实现一个微型shell 进程创建 在Linux系统中我们可以在一个进程使用系统调用fork()来创建子进程,创建出来的进程就是子进程,原来的进程为父进程。…...
鸿蒙DevEco Studio HarmonyOS 5跑酷小游戏实现指南
1. 项目概述 本跑酷小游戏基于鸿蒙HarmonyOS 5开发,使用DevEco Studio作为开发工具,采用Java语言实现,包含角色控制、障碍物生成和分数计算系统。 2. 项目结构 /src/main/java/com/example/runner/├── MainAbilitySlice.java // 主界…...
