一文弄明白KeyedProcessFunction函数

2024-02-23 20:20
文章标签 函数 一文 明白 keyedprocessfunction

本文主要是介绍一文弄明白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()}.*/

上面简单来说就是以下四点

  1. Flink中输入流中的每一条数据都会触发KeyedProcessFunction类的processElement方法调用
  2. 通过这个方法的Context参数可以设置定时器，在开启定时器后会程序会定时调用onTimer方法
  3. 由于KeyedProcessFunction实现了RichFunction接口，因此是可以通过RuntimeContext上下文对象管理状态state的开启和释放
  4. 需要注意的是，只有在KeyedStream里才能够访问state和定时器，通俗点来说就是这个函数要用在keyBy这个函数的后面

processElement方法解析

  1. Flink会调用processElement方法处理输入流中的每一条数据
  2. KeyedProcessFunction.Context参数可以用来读取以及更新内部状态state
  3. 这个KeyedProcessFunction跟其他function一样通过参数中的Collector对象以回写的方式返回数据

onTimer方法解析：在启用TimerService服务时会定时触发此方法，一般会在processElement方法中开启TimerService服务

以上就是这个函数的基本知识，接下来就通过实战来熟悉下它的使用

实战简介

本次实战的目标是学习KeyedProcessFunction，内容如下：

  1. 监听本机7777端口读取字符串
  2. 将每个字符串用空格分隔，转成Tuple2实例，f0是分隔后的单词，f1等于1
  3. 将Tuple2实例集合通过f0字段分区，得到KeyedStream
  4. KeyedSteam通过自定义KeyedProcessFunction处理
  5. 自定义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

  1. 启动Flink服务后，往7777端口里面发送数据
    在这里插入图片描述

  2. 通过IDEA的终端可以看到有日志输出，可以看到在发送消息的时候第一条日志立马打印出来并在10秒后输出第二条日志
    在这里插入图片描述

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

  4. 再输入点其他的试试
    在这里插入图片描述

  5. 通过输出可以看到这些单词的计数器又从0开始，说明每一个Key都对应一个状态
    在这里插入图片描述

思考题

  1. open方法会在哪里进行调用，KeyedProcessFunction整个类的完整调用逻辑是怎么样的
  2. registerProcessingTimeTimer和registerEventTimeTimer的差异是什么

参考资料

  1. https://blog.csdn.net/boling_cavalry/article/details/106299167
  2. https://blog.csdn.net/lujisen/article/details/105510532
  3. https://blog.csdn.net/qq_31866793/article/details/102831731

这篇关于一文弄明白KeyedProcessFunction函数的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

C++统计函数执行时间的最佳实践

C++统计函数执行时间的最佳实践

《C++统计函数执行时间的最佳实践》在软件开发过程中,性能分析是优化程序的重要环节,了解函数的执行时间分布对于识别性能瓶颈至关重要,本文将分享一个C++函数执行时间统计工具,希望对大家有所帮助... 目录前言工具特性核心设计1. 数据结构设计2. 单例模式管理器3. RAII自动计时使用方法基本用法高级用法
阅读更多...
GO语言中函数命名返回值的使用

GO语言中函数命名返回值的使用

《GO语言中函数命名返回值的使用》在Go语言中,函数可以为其返回值指定名称,这被称为命名返回值或命名返回参数,这种特性可以使代码更清晰,特别是在返回多个值时,感兴趣的可以了解一下... 目录基本语法函数命名返回特点代码示例命名特点基本语法func functionName(parameters) (nam
阅读更多...
Python Counter 函数使用案例

Python Counter 函数使用案例

《PythonCounter函数使用案例》Counter是collections模块中的一个类,专门用于对可迭代对象中的元素进行计数,接下来通过本文给大家介绍PythonCounter函数使用案例... 目录一、Counter函数概述二、基本使用案例（一）列表元素计数（二）字符串字符计数（三）元组计数三、C
阅读更多...
Python中的filter() 函数的工作原理及应用技巧

Python中的filter() 函数的工作原理及应用技巧

《Python中的filter()函数的工作原理及应用技巧》Python的filter()函数用于筛选序列元素,返回迭代器,适合函数式编程,相比列表推导式,内存更优,尤其适用于大数据集,结合lamb... 目录前言一、基本概念基本语法二、使用方式1. 使用 lambda 函数2. 使用普通函数3. 使用 N
阅读更多...
MySQL中REPLACE函数与语句举例详解

MySQL中REPLACE函数与语句举例详解

《MySQL中REPLACE函数与语句举例详解》在MySQL中REPLACE函数是一个用于处理字符串的强大工具,它的主要功能是替换字符串中的某些子字符串,：本文主要介绍MySQL中REPLACE函... 目录一、REPLACE()函数语法：参数说明：功能说明：示例：二、REPLACE INTO语句语法：参数
阅读更多...
python中update()函数的用法和一些例子

python中update()函数的用法和一些例子

《python中update()函数的用法和一些例子》update()方法是字典对象的方法,用于将一个字典中的键值对更新到另一个字典中,：本文主要介绍python中update()函数的用法和一些... 目录前言用法注意事项示例示例 1: 使用另一个字典来更新示例 2: 使用可迭代对象来更新示例 3: 使用
阅读更多...
一文带你迅速搞懂路由器/交换机/光猫三者概念区别

一文带你迅速搞懂路由器/交换机/光猫三者概念区别

《一文带你迅速搞懂路由器/交换机/光猫三者概念区别》讨论网络设备时，常提及路由器、交换机及光猫等词汇，日常生活、工作中，这些设备至关重要，居家上网、企业内部沟通乃至互联网冲浪皆无法脱离其影响力，本文将... 当谈论网络设备时，我们常常会听到路由器、交换机和光猫这几个名词。它们是构建现代网络基础设施的关键组成
阅读更多...
Python lambda函数(匿名函数)、参数类型与递归全解析

Python lambda函数(匿名函数)、参数类型与递归全解析

《Pythonlambda函数(匿名函数)、参数类型与递归全解析》本文详解Python中lambda匿名函数、灵活参数类型和递归函数三大进阶特性,分别介绍其定义、应用场景及注意事项,助力编写简洁高效... 目录一、lambda 匿名函数：简洁的单行函数1. lambda 的定义与基本用法2. lambda
阅读更多...
Python 函数详解:从基础语法到高级使用技巧

Python 函数详解:从基础语法到高级使用技巧

《Python函数详解:从基础语法到高级使用技巧》本文基于实例代码,全面讲解Python函数的定义、参数传递、变量作用域及类型标注等知识点,帮助初学者快速掌握函数的使用技巧,感兴趣的朋友跟随小编一起... 目录一、函数的基本概念与作用二、函数的定义与调用1. 无参函数2. 带参函数3. 带返回值的函数4.
阅读更多...
MySQL中DATE_FORMAT时间函数的使用小结

MySQL中DATE_FORMAT时间函数的使用小结

《MySQL中DATE_FORMAT时间函数的使用小结》本文主要介绍了MySQL中DATE_FORMAT时间函数的使用小结,用于格式化日期/时间字段,可提取年月、统计月份数据、精确到天,对大家的学习或... 目录前言DATE_FORMAT时间函数总结前言mysql可以使用DATE_FORMAT获取日期字段
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2