【Flink】Process Function

目录

1、ProcessFunction解析

1.1 抽象方法.processElement()

1.2 非抽象方法.onTimer()

2、Flink中8个不同的处理函数

2.1 ProcessFunction

2.2 KeyedProcessFunction

2.3 ProcessWindowFunction

2.4 ProcessAllWindowFunction

2.5 CoProcessFunction

2.6 ProcessJoinFunction

2.7 BroadcastProcessFunction

2.8 KeyedBroadcastProcessFunction

3、按键分区处理函数

3.1 定时器（Timer）和定时服务（TimerService）

4、窗口处理函数

4.1 窗口函数使用

4.2 ProcessWindowFunction解析

它是底层提炼的一个可以自定义处理逻辑的操作，被叫作“处理函数”（process function）。

1、ProcessFunction解析

在源码中我们可以看到，抽象类ProcessFunction继承了AbstractRichFunction，有两个泛型类型参数：I表示Input，也就是输入的数据类型；O表示Output，也就是处理完成之后输出的数据类型。

内部单独定义了两个方法：一个是必须要实现的抽象方法.processElement()；另一个是非抽象方法.onTimer()。

public abstract class ProcessFunction<I, O> extends AbstractRichFunction {...public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception;public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) throws Exception {}...}

1.1 抽象方法.processElement()

用于“处理元素”，定义了处理的核心逻辑。这个方法对于流中的每个元素都会调用一次，参数包括三个：输入数据值value，上下文ctx，以及“收集器”（Collector）out。方法没有返回值，处理之后的输出数据是通过收集器out来定义的。

  1. value：当前流中的输入元素，也就是正在处理的数据，类型与流中数据类型一致。
  2. ctx：类型是ProcessFunction中定义的内部抽象类Context，表示当前运行的上下文，可以获取到当前的时间戳，并提供了用于查询时间和注册定时器的“定时服务”（TimerService），以及可以将数据发送到“侧输出流”（side output）的方法.output()。
  3. out：“收集器”（类型为Collector），用于返回输出数据。使用方式与flatMap算子中的收集器完全一样，直接调用out.collect()方法就可以向下游发出一个数据。这个方法可以多次调用，也可以不调用。

通过几个参数的分析不难发现，ProcessFunction可以轻松实现flatMap、map、filter这样的基本转换功能；而通过富函数提供的获取上下文方法.getRuntimeContext()，也可以自定义状态（state）进行处理，这也就能实现聚合操作的功能了。

1.2 非抽象方法.onTimer()

定时方法.onTimer()也有三个参数：时间戳（timestamp），上下文（ctx），以及收集器（out）。这里的timestamp是指设定好的触发时间，事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器，所以也可以调用定时服务（TimerService），以及任意输出处理之后的数据。

注意：在Flink中，只有“按键分区流”KeyedStream才支持设置定时器的操作。

2、Flink中8个不同的处理函数

2.1 ProcessFunction

最基本的处理函数，基于DataStream直接调用.process()时作为参数传入。

2.2 KeyedProcessFunction

对流按键分区后的处理函数，基于KeyedStream调用.process()时作为参数传入。要想使用定时器，比如基于KeyedStream。

2.3 ProcessWindowFunction

开窗之后的处理函数，也是全窗口函数的代表。基于WindowedStream调用.process()时作为参数传入。

2.4 ProcessAllWindowFunction

同样是开窗之后的处理函数，基于AllWindowedStream调用.process()时作为参数传入。

2.5 CoProcessFunction

合并（connect）两条流之后的处理函数，基于ConnectedStreams调用.process()时作为参数传入。关于流的连接合并操作，我们会在后续章节详细介绍。

2.6 ProcessJoinFunction

间隔连接（interval join）两条流之后的处理函数，基于IntervalJoined调用.process()时作为参数传入。

2.7 BroadcastProcessFunction

广播连接流处理函数，基于BroadcastConnectedStream调用.process()时作为参数传入。这里的“广播连接流”BroadcastConnectedStream，是一个未keyBy的普通DataStream与一个广播流（BroadcastStream）做连接（conncet）之后的产物。关于广播流的相关操作，我们会在后续章节详细介绍。

2.8 KeyedBroadcastProcessFunction

按键分区的广播连接流处理函数，同样是基于BroadcastConnectedStream调用.process()时作为参数传入。与BroadcastProcessFunction不同的是，这时的广播连接流，是一个KeyedStream与广播流（BroadcastStream）做连接之后的产物。

3、按键分区处理函数

只有在KeyedStream中才支持使用TimerService设置定时器的操作。所以一般情况下，我们都是先做了keyBy分区之后，再去定义处理操作；代码中更加常见的处理函数是KeyedProcessFunction。

3.1 定时器（Timer）和定时服务（TimerService）

定时服务与当前运行的环境有关。ProcessFunction的上下文（Context）中提供了.timerService()方法，可以直接返回一个TimerService对象。TimerService是Flink关于时间和定时器的基础服务接口，包含以下六个方法：

// 获取当前的处理时间
long currentProcessingTime();// 获取当前的水位线（事件时间）
long currentWatermark();// 注册处理时间定时器，当处理时间超过time时触发
void registerProcessingTimeTimer(long time);// 注册事件时间定时器，当水位线超过time时触发
void registerEventTimeTimer(long time);// 删除触发时间为time的处理时间定时器
void deleteProcessingTimeTimer(long time);// 删除触发时间为time的处理时间定时器
void deleteEventTimeTimer(long time);

TimerService会以键（key）和时间戳为标准，对定时器进行去重；也就是说对于每个key和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。

public class KeyedProcessTimerDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 7777).map(new WaterSensorMapFunction()).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((element, ts) -> element.getTs() * 1000L));KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());// TODO Process:keyedSingleOutputStreamOperator<String> process = sensorKS.process(new KeyedProcessFunction<String, WaterSensor, String>() {/*** 来一条数据调用一次* @param value* @param ctx* @param out* @throws Exception*/@Overridepublic void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {//获取当前数据的keyString currentKey = ctx.getCurrentKey();// TODO 1.定时器注册TimerService timerService = ctx.timerService();// 1、事件时间的案例Long currentEventTime = ctx.timestamp(); // 数据中提取出来的事件时间timerService.registerEventTimeTimer(5000L);System.out.println("当前key=" + currentKey + ",当前时间=" + currentEventTime + ",注册了一个5s的定时器");// 2、处理时间的案例
//                        long currentTs = timerService.currentProcessingTime();
//                        timerService.registerProcessingTimeTimer(currentTs + 5000L);
//                        System.out.println("当前key=" + currentKey + ",当前时间=" + currentTs + ",注册了一个5s后的定时器");// 3、获取 process的 当前watermark
//                        long currentWatermark = timerService.currentWatermark();
//                        System.out.println("当前数据=" + value + ",当前watermark=" + currentWatermark);// 注册定时器： 处理时间、事件时间
//                        timerService.registerProcessingTimeTimer();
//                        timerService.registerEventTimeTimer();// 删除定时器： 处理时间、事件时间
//                        timerService.deleteEventTimeTimer();
//                        timerService.deleteProcessingTimeTimer();// 获取当前时间进展： 处理时间-当前系统时间，  事件时间-当前watermark
//                        long currentTs = timerService.currentProcessingTime();
//                        long wm = timerService.currentWatermark();}/*** TODO 2.时间进展到定时器注册的时间，调用该方法* @param timestamp 当前时间进展，就是定时器被触发时的时间* @param ctx       上下文* @param out       采集器* @throws Exception*/@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {super.onTimer(timestamp, ctx, out);String currentKey = ctx.getCurrentKey();System.out.println("key=" + currentKey + "现在时间是" + timestamp + "定时器触发");}});process.print();env.execute();}
}

4、窗口处理函数

除了KeyedProcessFunction，另外一大类常用的处理函数，就是基于窗口的ProcessWindowFunction和ProcessAllWindowFunction了。

4.1 窗口函数使用

stream.keyBy( t -> t.f0 ).window( TumblingEventTimeWindows.of(Time.seconds(10)) ).process(new MyProcessWindowFunction())

4.2 ProcessWindowFunction解析

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> extends AbstractRichFunction {...public abstract void process(KEY key, Context context, Iterable<IN> elements, Collector<OUT> out) throws Exception;public void clear(Context context) throws Exception {}public abstract class Context implements java.io.Serializable {...}
}

ProcessWindowFunction依然是一个继承了AbstractRichFunction的抽象类，它有四个类型参数：

  • IN：input，数据流中窗口任务的输入数据类型。
  • OUT：output，窗口任务进行计算之后的输出数据类型。
  • KEY：数据中键key的类型。
  • W：窗口的类型，是Window的子类型。一般情况下我们定义时间窗口，W就是TimeWindow。

ProcessWindowFunction里面处理数据的核心方法.process()。方法包含四个参数。

  • key：窗口做统计计算基于的键，也就是之前keyBy用来分区的字段。
  • context：当前窗口进行计算的上下文，它的类型就是ProcessWindowFunction内部定义的抽象类Context。
  • elements：窗口收集到用来计算的所有数据，这是一个可迭代的集合类型。
  • out：用来发送数据输出计算结果的收集器，类型为Collector。

可以明显看出，这里的参数不再是一个输入数据，而是窗口中所有数据的集合。而上下文context所包含的内容也跟其他处理函数有所差别：

public abstract class Context implements java.io.Serializable {public abstract W window();public abstract long currentProcessingTime();public abstract long currentWatermark();public abstract KeyedStateStore windowState();public abstract KeyedStateStore globalState();public abstract <X> void output(OutputTag<X> outputTag, X value);}

除了可以通过.output()方法定义侧输出流不变外，其他部分都有所变化。这里不再持有TimerService对象，只能通过currentProcessingTime()和currentWatermark()来获取当前时间，所以失去了设置定时器的功能；另外由于当前不是只处理一个数据，所以也不再提供.timestamp()方法。与此同时，也增加了一些获取其他信息的方法：比如可以通过.window()直接获取到当前的窗口对象，也可以通过.windowState()和.globalState()获取到当前自定义的窗口状态和全局状态。注意这里的“窗口状态”是自定义的，不包括窗口本身已经有的状态，针对当前key、当前窗口有效；而“全局状态”同样是自定义的状态，针对当前key的所有窗口有效。

所以我们会发现，ProcessWindowFunction中除了.process()方法外，并没有.onTimer()方法，而是多出了一个.clear()方法。从名字就可以看出，这主要是方便我们进行窗口的清理工作。如果我们自定义了窗口状态，那么必须在.clear()方法中进行显式地清除，避免内存溢出。

至于另一种窗口处理函数ProcessAllWindowFunction，它的用法非常类似。区别在于它基于的是AllWindowedStream，相当于对没有keyBy的数据流直接开窗并调用.process()方法：

stream.windowAll( TumblingEventTimeWindows.of(Time.seconds(10)) ).process(new MyProcessAllWindowFunction())

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/156881.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

https和http的区别和优势

https和http的区别和优势

大家好&#xff0c;我是咕噜-凯撒&#xff0c;HTTP&#xff08;超文本传输协议&#xff09;和HTTPS&#xff08;安全超文本传输协议&#xff09;是用于在网络上传输数据的协议&#xff0c;HTTPS相比HTTP在数据传输过程中更加安全可靠&#xff0c;适合对数据安全性要求较高的场景…
阅读更多...
孟德尔随机化写作技巧mr

孟德尔随机化写作技巧mr

阅读更多...
ventoy安装操作系统

ventoy安装操作系统

下载ventoy https://github.com/ventoy/Ventoy/releases/download/v1.0.96/ventoy-1.0.96-windows.zip 解压后执行 Ventoy2Disk 2、安装后将ISO放入U盘大的分区&#xff0c;通过U盘启动就可以识别到ISO镜像开始装系统
阅读更多...
MySQL 日志管理、备份与恢复

MySQL 日志管理、备份与恢复

一、MySQL 日志管理 MySQL 的日志默认保存位置为 /usr/local/mysql/data vim /etc/my.cnf [mysqld] ##错误日志&#xff0c;用来记录当MySQL启动、停止或运行时发生的错误信息&#xff0c;默认已开启 log-error/usr/local/mysql/data/mysql_error.log #指定日志的保存位置…
阅读更多...
springboot项目基于jdk17、分布式事务seata-server-1.7.1、分库分表shardingSphere5.2.1开发过程中出现的问题

springboot项目基于jdk17、分布式事务seata-server-1.7.1、分库分表shardingSphere5.2.1开发过程中出现的问题

由于项目需要&#xff0c;springboot项目需基于jdk17环境开发&#xff0c;结合nacos2.0.3、分布式事务seata-server-1.7.1、分库分表shardingSphere5.2.1等&#xff0c;项目启动过程中出现的问题解决方式小结。 问题一&#xff1a; Caused by: java.lang.RuntimeException: j…
阅读更多...
职场Excel：求和家族，不简单

职场Excel：求和家族，不简单

说到excel函数&#xff0c;很多人第一时间想到的就是求和函数sum。作为excel入门级函数&#xff0c;sum的确是小白级的&#xff0c;以至于很多人对求和函数有点“误解”&#xff0c;觉得求和函数太简单了。 但是&#xff0c;你可能不知道&#xff0c;sum只是excel求和家族里的一…
阅读更多...
Ubuntu22.04 交叉编译GCC13.2.0 for Rv1126

Ubuntu22.04 交叉编译GCC13.2.0 for Rv1126

一、安装Ubuntu22.04 sudo apt install vim net-tools openssh-server 二、安装必要项 sudo apt update sudo apt upgrade sudo apt install build-essential gawk git texinfo bison flex 三、下载必备软件包 1.glibc https://ftp.gnu.org/gnu/glibc/glibc-2.38.tar.gz…
阅读更多...
引迈-JNPF低代码项目技术栈介绍

引迈-JNPF低代码项目技术栈介绍

从 2014 开始研发低代码前端渲染&#xff0c;到 2018 年开始研发后端低代码数据模型&#xff0c;发布了JNPF开发平台。 谨以此文针对 JNPF-JAVA-Cloud微服务 进行相关技术栈展示&#xff1a; 1. 项目前后端分离 前端采用Vue.js&#xff0c;这是一种流行的前端JavaScript框架&a…
阅读更多...
SQLServer 数据库语句可以执行但是语句会有红线提示错误

SQLServer 数据库语句可以执行但是语句会有红线提示错误

刷新缓存即可&#xff0c;代码没有错误。
阅读更多...
4D毫米波雷达和3D雷达、激光雷达全面对比

4D毫米波雷达和3D雷达、激光雷达全面对比

众所周知&#xff0c;传统3D毫米波雷达存在如下性能缺陷&#xff1a; 1&#xff09;静止目标和地物杂波混在一起&#xff0c;难以区分&#xff1b; 2) 横穿车辆和行人多普勒为零或很低&#xff0c;难以检测&#xff1b; 3) 高处物体和地面目标不能区分&#xff0c;容易造成误刹…
阅读更多...
从CNN到Transformer：基于PyTorch的遥感影像、无人机影像的地物分类、目标检测、语义分割和点云分类

从CNN到Transformer：基于PyTorch的遥感影像、无人机影像的地物分类、目标检测、语义分割和点云分类

我国高分辨率对地观测系统重大专项已全面启动&#xff0c;高空间、高光谱、高时间分辨率和宽地面覆盖于一体的全球天空地一体化立体对地观测网逐步形成&#xff0c;将成为保障国家安全的基础性和战略性资源。随着小卫星星座的普及&#xff0c;对地观测已具备多次以上的全球覆盖…
阅读更多...
部署单仓库多目录项目

部署单仓库多目录项目

部署单仓库多目录项目 文章目录 部署单仓库多目录项目1.部署单仓库多目录项目2.Shell脚本进行部署单仓库多目录项目2.1 编写Shell脚本2.2 Demo推送代码及测试 3.小结 1.部署单仓库多目录项目 #部署单仓库多目录项目 在开发过程中,研发团队往往会将一个大型项目拆分成几个子目录…
阅读更多...
MCU 的 TOP 15 图形GUI库：选择最适合你的图形用户界面（一）

MCU 的 TOP 15 图形GUI库：选择最适合你的图形用户界面（一）

在嵌入式系统开发中&#xff0c;选择一个合适的图形用户界面&#xff08;GUI&#xff09;库是至关重要的。在屏幕上显示的时候&#xff0c;使用现成的图形库&#xff0c;这样开发人员就不需要弄清楚底层任务&#xff0c;例如如何绘制像素、线条、形状&#xff0c;如果再高级一点…
阅读更多...
基于骑手优化算法优化概率神经网络PNN的分类预测 - 附代码

基于骑手优化算法优化概率神经网络PNN的分类预测 - 附代码

基于骑手优化算法优化概率神经网络PNN的分类预测 - 附代码 文章目录 基于骑手优化算法优化概率神经网络PNN的分类预测 - 附代码1.PNN网络概述2.变压器故障诊街系统相关背景2.1 模型建立 3.基于骑手优化优化的PNN网络5.测试结果6.参考文献7.Matlab代码 摘要&#xff1a;针对PNN神…
阅读更多...
2021秋招-总目录

2021秋招-总目录

2021秋招-目录 知识点总结 预训练语言模型: Bert家族 1.1 BERT、attention、transformer理解部分 B站讲解–强烈推荐可视化推倒结合代码理解代码部分常见面试考点以及问题: word2vec 、 fasttext 、elmo;BN 、LN、CN、WNNLP中的loss与评价总结 4.1 loss_function&#xff1…
阅读更多...
基于AVR单片机的移动目标视觉追踪系统设计与实现

基于AVR单片机的移动目标视觉追踪系统设计与实现

基于AVR单片机的移动目标视觉追踪系统是一种常见的应用领域&#xff0c;旨在通过摄像头采集图像数据并使用图像处理和追踪算法实现对移动目标的实时追踪。本文将介绍基于AVR单片机的移动目标视觉追踪系统的设计原理和实现步骤&#xff0c;并提供相应的代码示例。 1. 设计概述 …
阅读更多...
基于SSM的校内互助交易平台设计

基于SSM的校内互助交易平台设计

末尾获取源码 开发语言&#xff1a;Java Java开发工具&#xff1a;JDK1.8 后端框架&#xff1a;SSM 前端&#xff1a;Vue 数据库&#xff1a;MySQL5.7和Navicat管理工具结合 服务器&#xff1a;Tomcat8.5 开发软件&#xff1a;IDEA / Eclipse 是否Maven项目&#xff1a;是 目录…
阅读更多...
MAX/MSP SDK学习04：Messages selector的使用

MAX/MSP SDK学习04：Messages selector的使用

其实消息选择器在simplemax示例中就接触到了&#xff0c;但这文档非要讲那么抽象。目前为止对消息选择器的理解是&#xff1a;可判断接收过来的消息是否符合本Object的处理要求&#xff0c;比如加法对象只可接收数值型的消息以处理&#xff0c;但不能接收t_symbol型的消息&…
阅读更多...
【Spring Boot】如何在Linux系统中快速启动Spring Boot的jar包

【Spring Boot】如何在Linux系统中快速启动Spring Boot的jar包

在Linux系统中先安装java的JDK 然后编写下列service.sh脚本&#xff0c;并根据自己的需求只需要修改export的log_path、exec_cmd参数即可 # 配置运行日志输出的路径 export log_path/usr/local/project/study-pro/logs # 当前服务运行的脚本命令 export exec_cmd"nohup /u…
阅读更多...
【DevOps】Git 图文详解（七）：标签管理

【DevOps】Git 图文详解（七）：标签管理

Git 图文详解&#xff08;七&#xff09;&#xff1a;标签管理 标签&#xff08;Tags&#xff09;指的是某个分支某个特定时间点的状态&#xff0c;是对某一个提交记录的 固定 “指针” 引用。一经创建&#xff0c;不可移动&#xff0c;存储在工作区根目录下 .git\refs\tags。可…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023