ProcessFunction:Flink最底层API使用踩坑记录

2024-09-06 20:58

本文主要是介绍ProcessFunction:Flink最底层API使用踩坑记录,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

大数据技术与架构

点击右侧关注,大数据开发领域最强公众号!

暴走大数据

点击右侧关注,暴走大数据!

ProcessFunction和CoProcessFunction

说明

DataStream与KeyedStreamd都有Process方法,
DataStream接收的是ProcessFunction,而KeyedStream接收的是KeyedProcessFunction(原本也支持ProcessFunction,现在已被废弃)

0.AbstractRichFunction介绍

1.ProcessFunction对flink更精细的操作

<1> Events(流中的事件)
<2> State(容错,一致性,仅仅用于keyed stream)
<3> Timers(事件时间和处理时间,仅仅适用于keyed stream)

ProcessFunction可以视为是FlatMapFunction,但是它可以获取keyed state和timers。每次有事件流入processFunction算子就会触发处理。

为了容错,ProcessFunction可以使用RuntimeContext访问flink内部的keyed state。

timer允许应用程序对处理时间和事件时间的变化做出反应。每次有事件到达都会调用函数processElement(...),该函数有参数,也就是Context对象,该对象可以访问元素的事件时间戳和TimerService,还有侧输出。

TimerService可用于注册为后续处理事件或者事件时间的回调。当达到计时器的特定时间时,将调用onTimer(...)方法。在该调用期间,所有状态再次限定为创建计时器的key,允许计时器操纵keyed状态。

2.CoProcessFunction 实现底层join

<1> 实现底层join操作典型模板就是:

  1. 为一个或者两个输入创建一个状态对象

  2. 根据输入的事件更新状态

  3. 根据从另一个流接受的元素,更新状态并且产生joined结果


3.KeyedProcessFunction

keyedProcessFunction是ProcessFunction的扩展,可以在onTimer获取timer的key (通过context.getCurrentKey方法)

4.Timer类型

1.两种类型(事件时间和处理时间)的timer都是由TimerService维护并且以队列的形式执行。

TimerService会使用key和timestamp对timer进行去重,也即是对于每一对key和timestamp仅仅会存在一个timer。如果同一个timestamp注册了多个timers,onTimer()函数仅仅会调用一次。

对于onTimer()和processElement()方法flink是做了同步的,所以不需要关系并发问题。


5.ProcessFunction与状态的结合使用案例

WordCount,如果某一个key一分钟(事件时间)没有更新,就直接输出。
基本思路:
// 1.ValueState内部包含了计数、key和最后修改时间
// 2.对于每一个输入的记录,ProcessFunction都会增加计数,并且修改时间戳
// 3.该函数会在事件时间的后续1min调度回调函数
// 4.然后根据每次回调函数,就去检查回调事件时间戳和保存的时间戳,如果匹配就将数据发出

public class ProcessFunctionExample {// 1.ValueState内部包含了计数、key和最后修改时间// 2.对于每一个输入的记录,ProcessFunction都会增加计数,并且修改时间戳// 3.该函数会在事件时间的后续1min调度回调函数// 4.然后根据每次回调函数,就去检查回调事件时间戳和保存的时间戳,如果匹配就将数据发出private static class StreamDataSource extends RichParallelSourceFunction<Tuple3<String, Long, Long>> {private volatile boolean running = true;@Overridepublic void run(SourceContext<Tuple3<String, Long, Long>> sourceContext) throws Exception {Tuple3[] elements = new Tuple3[]{Tuple3.of("a", 1L, 1000000050000L),Tuple3.of("a", 1L, 1000000054000L),Tuple3.of("a", 1L, 1000000079900L),Tuple3.of("a", 1L, 1000000115000L),Tuple3.of("b", 1L, 1000000100000L),Tuple3.of("b", 1L, 1000000108000L)};int count = 0;while (running && count < elements.length) {sourceContext.collect(new Tuple3<>((String) elements[count].f0, (Long) elements[count].f1, (Long) elements[count].f2));count++;Thread.sleep(10000);}}@Overridepublic void cancel() {running = false;}}/*** 存储在状态中的对象*/public static class CountWithTimestamp {//单词public String key;//单词计数public long count;//最近更新时间public long lastModified;@Overridepublic String toString() {return "CountWithTimestamp{" +"key='" + key + '\'' +", count=" + count +", lastModified=" + new Date(lastModified) +'}';}}/*** ProcessFunction有两个泛型类,一个输入一个输出*/public static class CountWithTimeoutFunction extends ProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>> {private ValueState<CountWithTimestamp> state;//最先调用@Overridepublic void open(Configuration parameters) throws Exception {//根据上下文获取状态state = getRuntimeContext().getState(new ValueStateDescriptor<CountWithTimestamp>("myState", CountWithTimestamp.class));}@Overridepublic void processElement(Tuple2<String, Long> input, Context context, Collector<Tuple2<String, Long>> output) throws Exception {CountWithTimestamp current = state.value();if (current == null) {current = new CountWithTimestamp();current.key = input.f0;}//更新ValueStatecurrent.count++;//这里面的context可以获取时间戳//todo 此时这里的时间戳可能为null,如果设置的时间为ProcessingTimecurrent.lastModified = context.timestamp();System.out.println("元素"+input.f0+"进入事件时间为:" + new Date(current.lastModified));state.update(current);//注册ProcessTimer,更新一次就会有一个ProcessTimercontext.timerService().registerEventTimeTimer(current.lastModified + 9000);System.out.println("定时触发时间为:"+new Date(current.lastModified + 9000));}//EventTimer被触发后产生的行为//todo 这里的timestamp是触发时间@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out) throws Exception {//获取上次时间,与参数中的timestamp相比,如果相差等于60s 就会输出CountWithTimestamp res = state.value();System.out.println("当前时间为:"+new Date(timestamp)+res);if (timestamp >= res.lastModified + 9000) {System.out.println("定时器被触发:"+"当前时间为"+new Date(timestamp)+" 最近修改时间为"+new Date(res.lastModified));out.collect(new Tuple2<String, Long>(res.key, res.count));}}}//执行主类public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<Tuple2<String, Long>> data = env.addSource(new StreamDataSource()).setParallelism(1).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, Long>>(Time.milliseconds(0)) {@Overridepublic long extractTimestamp(Tuple3<String, Long, Long> input) {return input.f2;}}).map(new MapFunction<Tuple3<String, Long, Long>, Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> map(Tuple3<String, Long, Long> input) throws Exception {return new Tuple2<>(input.f0, input.f1);}});data.keyBy(0).process(new CountWithTimeoutFunction()).print();env.execute();}}

这一步的结果是:

发现共有四个OnTimer被执行,其中没有执行OnTimer的两条元素是

这两条消息定时器预计执行时间都超过了09:48:35,因为这个案例采用的是事件时间,而这六条元素最大的事件时间为09:48:35,所以默认到09:48:35就停止了

注意:看代码可以发现这里发送的元素之间是每隔10秒发送,因为以为会影响结果,实际是我们使用的是EventTime,所以OnTimer被执行的时间,是看事件时间。

如果将最大事件时间改一下,改成


结果就是除了他自身,其余onTimer全部被执行了,因为它的事件时间,超过了其余5个元素的定时器触发时间。

并且我们发现有一条消息满足了其中的条件。

这里有一个疑问就是:为什么a的所有最近修改时间都是09:48:45 ,a的最大事件时间????
分析可能是构造的数据源的原因。这里模拟的是将优先数据源作为无限数据源使用

解决问题:

一开始没有设置为EventTime,所以在处理的时候还是以Process Time来处理的。
改完之后的效果:

分析问题产生的原因:因为一开始未指定时间类型为EventTime,所以默认是以Process Time来处理,而一般来说使用ProcessTime,就不需要指定Watermark了(Watermark只是与EventTime配合使用),但是代码中偏偏还是使用了assign...方法,所以会在数据加载完了,使用最近的元素的时间,生成一个Watermark,这时候有了Watermark才会执行onTimer方法,所以才会出现数据全部加载完,才执行onTimer方法;

而当指定为EventTime时,来一个元素就会生成一个Watermark,当Watermark大于某个元素的触发时间,OnTimer就会执行,而不是等数据全部加载完之后才会生成

所以上面一开始对某些onTimer没有执行的理解是错误的,应该按照上面没有指定EventTime的方式去理解。

欢迎点赞+收藏+转发朋友圈素质三连

文章不错?点个【在看】吧! ????

这篇关于ProcessFunction:Flink最底层API使用踩坑记录的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1143098

相关文章

Python 字典 (Dictionary)使用详解

《Python字典(Dictionary)使用详解》字典是python中最重要,最常用的数据结构之一,它提供了高效的键值对存储和查找能力,:本文主要介绍Python字典(Dictionary)... 目录字典1.基本特性2.创建字典3.访问元素4.修改字典5.删除元素6.字典遍历7.字典的高级特性默认字典

Zabbix在MySQL性能监控方面的运用及最佳实践记录

《Zabbix在MySQL性能监控方面的运用及最佳实践记录》Zabbix通过自定义脚本和内置模板监控MySQL核心指标(连接、查询、资源、复制),支持自动发现多实例及告警通知,结合可视化仪表盘,可有效... 目录一、核心监控指标及配置1. 关键监控指标示例2. 配置方法二、自动发现与多实例管理1. 实践步骤

使用Python构建一个高效的日志处理系统

《使用Python构建一个高效的日志处理系统》这篇文章主要为大家详细讲解了如何使用Python开发一个专业的日志分析工具,能够自动化处理、分析和可视化各类日志文件,大幅提升运维效率,需要的可以了解下... 目录环境准备工具功能概述完整代码实现代码深度解析1. 类设计与初始化2. 日志解析核心逻辑3. 文件处

一文详解如何使用Java获取PDF页面信息

《一文详解如何使用Java获取PDF页面信息》了解PDF页面属性是我们在处理文档、内容提取、打印设置或页面重组等任务时不可或缺的一环,下面我们就来看看如何使用Java语言获取这些信息吧... 目录引言一、安装和引入PDF处理库引入依赖二、获取 PDF 页数三、获取页面尺寸(宽高)四、获取页面旋转角度五、判断

C++中assign函数的使用

《C++中assign函数的使用》在C++标准模板库中,std::list等容器都提供了assign成员函数,它比操作符更灵活,支持多种初始化方式,下面就来介绍一下assign的用法,具有一定的参考价... 目录​1.assign的基本功能​​语法​2. 具体用法示例​​​(1) 填充n个相同值​​(2)

Spring StateMachine实现状态机使用示例详解

《SpringStateMachine实现状态机使用示例详解》本文介绍SpringStateMachine实现状态机的步骤,包括依赖导入、枚举定义、状态转移规则配置、上下文管理及服务调用示例,重点解... 目录什么是状态机使用示例什么是状态机状态机是计算机科学中的​​核心建模工具​​,用于描述对象在其生命

使用Python删除Excel中的行列和单元格示例详解

《使用Python删除Excel中的行列和单元格示例详解》在处理Excel数据时,删除不需要的行、列或单元格是一项常见且必要的操作,本文将使用Python脚本实现对Excel表格的高效自动化处理,感兴... 目录开发环境准备使用 python 删除 Excphpel 表格中的行删除特定行删除空白行删除含指定

深入理解Go语言中二维切片的使用

《深入理解Go语言中二维切片的使用》本文深入讲解了Go语言中二维切片的概念与应用,用于表示矩阵、表格等二维数据结构,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习吧... 目录引言二维切片的基本概念定义创建二维切片二维切片的操作访问元素修改元素遍历二维切片二维切片的动态调整追加行动态

prometheus如何使用pushgateway监控网路丢包

《prometheus如何使用pushgateway监控网路丢包》:本文主要介绍prometheus如何使用pushgateway监控网路丢包问题,具有很好的参考价值,希望对大家有所帮助,如有错误... 目录监控网路丢包脚本数据图表总结监控网路丢包脚本[root@gtcq-gt-monitor-prome

Python通用唯一标识符模块uuid使用案例详解

《Python通用唯一标识符模块uuid使用案例详解》Pythonuuid模块用于生成128位全局唯一标识符,支持UUID1-5版本,适用于分布式系统、数据库主键等场景,需注意隐私、碰撞概率及存储优... 目录简介核心功能1. UUID版本2. UUID属性3. 命名空间使用场景1. 生成唯一标识符2. 数