一文弄明白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。

相关文章

hdu1171（母函数或多重背包）

hdu1171（母函数或多重背包）

题意：把物品分成两份，使得价值最接近 可以用背包，或者是母函数来解，母函数（1 + x^v+x^2v+.....+x^num*v）（1 + x^v+x^2v+.....+x^num*v）（1 + x^v+x^2v+.....+x^num*v） 其中指数为价值，每一项的数目为（该物品数+1）个 代码如下： #include<iostream>#include<algorithm>
阅读更多...
C++操作符重载实例(独立函数)

C++操作符重载实例(独立函数)

C++操作符重载实例，我们把坐标值CVector的加法进行重载，计算c3=c1+c2时，也就是计算x3=x1+x2，y3=y1+y2，今天我们以独立函数的方式重载操作符+（加号），以下是C++代码： c1802.cpp源代码： D:\YcjWork\CppTour>vim c1802.cpp #include <iostream>using namespace std;/*** 以独立函数
阅读更多...
函数式编程思想

函数式编程思想

我们经常会用到各种各样的编程思想，例如面向过程、面向对象。不过笔者在该博客简单介绍一下函数式编程思想. 如果对函数式编程思想进行概括，就是f(x) = na(x) , y=uf(x)…至于其他的编程思想，可能是y=a(x)+b(x)+c(x)…，也有可能是y=f(x)=f(x)/a + f(x)/b+f(x)/c… 面向过程的指令式编程 面向过程，简单理解就是y=a(x)+b(x)+c(x)
阅读更多...
利用matlab bar函数绘制较为复杂的柱状图，并在图中进行适当标注

利用matlab bar函数绘制较为复杂的柱状图，并在图中进行适当标注

示例代码和结果如下：小疑问：如何自动选择合适的坐标位置对柱状图的数值大小进行标注？😂 clear; close all;x = 1:3;aa=[28.6321521955954 26.2453660695847 21.69102348512086.93747104431360 6.25442246899816 3.342835958564245.51365061796319 4.87
阅读更多...
OpenCV结构分析与形状描述符（11）椭圆拟合函数fitEllipse()的使用

OpenCV结构分析与形状描述符（11）椭圆拟合函数fitEllipse()的使用

操作系统：ubuntu22.04 OpenCV版本：OpenCV4.9 IDE:Visual Studio Code 编程语言：C++11 算法描述 围绕一组2D点拟合一个椭圆。 该函数计算出一个椭圆，该椭圆在最小二乘意义上最好地拟合一组2D点。它返回一个内切椭圆的旋转矩形。使用了由[90]描述的第一个算法。开发者应该注意，由于数据点靠近包含的 Mat 元素的边界，返回的椭圆/旋转矩形数据
阅读更多...
Unity3D 运动之Move函数和translate

Unity3D 运动之Move函数和translate

CharacterController.Move 移动 function Move (motion : Vector3) : CollisionFlags Description描述 A more complex move function taking absolute movement deltas. 一个更加复杂的运动函数，每次都绝对运动。 Attempts to
阅读更多...
✨机器学习笔记（二）—— 线性回归、代价函数、梯度下降

✨机器学习笔记（二）—— 线性回归、代价函数、梯度下降

1️⃣线性回归（linear regression） f w , b ( x ) = w x + b f_{w,b}(x) = wx + b fw,b​(x)=wx+b 🎈A linear regression model predicting house prices： 如图是机器学习通过监督学习运用线性回归模型来预测房价的例子，当房屋大小为1250 f e e t 2 feet^
阅读更多...
JavaSE（十三）——函数式编程（Lambda表达式、方法引用、Stream流）

JavaSE（十三）——函数式编程（Lambda表达式、方法引用、Stream流）

函数式编程 函数式编程 是 Java 8 引入的一个重要特性，它允许开发者以函数作为一等公民（first-class citizens）的方式编程，即函数可以作为参数传递给其他函数，也可以作为返回值。 这极大地提高了代码的可读性、可维护性和复用性。函数式编程的核心概念包括高阶函数、Lambda 表达式、函数式接口、流（Streams）和 Optional 类等。 函数式编程的核心是Lambda
阅读更多...
PHP APC缓存函数使用教程

PHP APC缓存函数使用教程

APC，全称是Alternative PHP Cache，官方翻译叫”可选PHP缓存”。它为我们提供了缓存和优化PHP的中间代码的框架。 APC的缓存分两部分:系统缓存和用户数据缓存。(Linux APC扩展安装) 系统缓存 它是指APC把PHP文件源码的编译结果缓存起来，然后在每次调用时先对比时间标记。如果未过期，则使用缓存的中间代码运行。默认缓存 3600s(一小时)。但是这样仍会浪费大量C
阅读更多...
PHP7扩展开发之函数方式使用lib库

PHP7扩展开发之函数方式使用lib库

前言 首先说下什么是lib库。lib库就是一个提供特定功能的一个文件。可以把它看成是PHP的一个文件，这个文件提供一些函数方法。只是这个lib库是用c或者c++写的。 使用lib库的场景。一些软件已经提供了lib库，我们就没必要再重复实现一次。如，原先的mysql扩展，就是使用mysql官方的lib库进行的封装。 在本文，我们将建立一个简单的lib库，并在扩展中进行封装调用。 代码 基础
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2