结合案例总结Flink框架中的最底层API（ProcessFunction)用法 _function

书史足自悦，安用勤与劬。这篇文章主要讲述结合案例总结Flink框架中的最底层API（ProcessFunction)用法相关的知识，希望能为你提供帮助。

概述在之前总结的文章中有提到过，Flink框架提供了三层API完成流处理任务。至此已经学习了DataStream API ，ProcessFunction API 是Flink中最底层的API，可以访问时间戳、watermark 以及注册定时事件。还可以输出特定的一些事件。、

文章图片

【结合案例总结Flink框架中的最底层API（ProcessFunction)用法】Process Function 用来构建事件驱动的应用以及实现自定义的业务逻辑，若窗口函数以及转换算子都无法满足业务的要求时，需要请出ProcessFunction 去完成开发任务。Flink SQL 就是使用 Process Function 实现的。
Flink 提供了 8 个 Process Function如下：??ProcessFunction???、??KeyedProcessFunction???、??CoProcessFunction???、??ProcessJoinFunction???、??BroadcastProcessFunction???、??KeyedBroadcastProcessFunction???、??ProcessWindowFunction???、??ProcessAllWindowFunction??。接下来我们以KeyedProcessFunction为例来进行学习。
KeyedProcessFunction< K, I, O> 它主要用来操作KeyedStream，会处理流的每一个元素，输出为 0 个、1 个或者多个元素。所有的 Process Function 都继承自RichFunction 接口，所以都有 open()、close()和 getRuntimeContext()等方法。除此之外还提供了两个方法：

数据流中的每一个元素都会调用这个方法，调用结果将会放在 Collector 数据类型中输出。Context 可以访问元素的时间戳，元素的key，以及 TimerService 时间服务。Context 还可以将结果输出到别的流(side outputs)。

processElement(I& nbsp; value,& nbsp; Context& nbsp; ctx,& nbsp; Collector& lt; O& gt; & nbsp; out)

当之前注册的定时器触发时调用。参数 timestamp 为定时器所设定的触发的时间戳。Collector为输出结果的集合。OnTimerContext 和processElement 的 Context 参数一样，提供了上下文的一些信息。例如定时器触发的时间信息(事件时间或者处理时间)。

onTimer(long& nbsp; timestamp,& nbsp; OnTimerContext& nbsp; ctx,& nbsp; Collector& lt; O& gt; & nbsp; out)

定时器Context 中 TimerService对象方汇总：

返回当前处理时间

long& nbsp; currentProcessingTime()

返回当前 watermark 的时间戳

long& nbsp; currentWatermark()

注册当前 key 的processing time 的定时器，当 processing time 到达定时时间时，触发 timer。

void& nbsp; registerProcessingTimeTimer(long& nbsp; timestamp)

注册当前 key 的 event time 定时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。

void& nbsp; registerEventTimeTimer(long& nbsp; timestamp)

删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行

void& nbsp; deleteProcessingTimeTimer(long& nbsp; timestamp)

删除之前注册的事件时间定时器，如果没有此时间戳的定时器，则不执行。

void& nbsp; deleteEventTimeTimer(long& nbsp; timestamp)

测试代码：

public& nbsp; class& nbsp; ProcessTest1_KeyedProcessFunction& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; static& nbsp; void& nbsp; main(String[]& nbsp; args)& nbsp; throws& nbsp; Exception
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; StreamExecutionEnvironment& nbsp; env& nbsp; =& nbsp; StreamExecutionEnvironment.getExecutionEnvironment();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; env.setParallelism(1);

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; DataStreamSource& lt; String& gt; & nbsp; inputStream& nbsp; =& nbsp; env.socketTextStream(& quot; localhost& quot; ,& nbsp; 7777);
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; DataStream& lt; SensorReading& gt; & nbsp; dataStream& nbsp; =& nbsp; inputStream.map(line& nbsp; -& gt; & nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; String[]& nbsp; fields& nbsp; =& nbsp; line.split(& quot; ,& quot; );
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; return& nbsp; new& nbsp; SensorReading(fields[0],& nbsp; new& nbsp; Long(fields[1]),& nbsp; new& nbsp; Double(fields[2]));
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; );

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //测试keyedprocessFunction 先分组，自定义处理
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; dataStream.keyBy(& quot; id& quot; )
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; .process(new& nbsp; MyProcess())
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; .print();

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; env.execute();
& nbsp; & nbsp; & nbsp; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; // 实现自定义处理函数
& nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; static& nbsp; class& nbsp; MyProcess& nbsp; extends& nbsp; KeyedProcessFunction& lt; Tuple,SensorReading,Integer& gt; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ValueState& lt; Long& gt; & nbsp; tsTimeState;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; @Override
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; void& nbsp; open(Configuration& nbsp; parameters)& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; tsTimeState& nbsp; =& nbsp; getRuntimeContext().getState(new& nbsp; ValueStateDescriptor& lt; Long& gt; (
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & quot; tsTimeState& quot; ,Long.class
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ));
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; @Override
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; void& nbsp; processElement(SensorReading& nbsp; value,& nbsp; Context& nbsp; ctx,& nbsp; Collector& lt; Integer& gt; & nbsp; out)& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; out.collect(value.getId().length());

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; // Context操作
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timestamp();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.getCurrentKey();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; // 侧流
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //ctx.output();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; // 获取当前系统处理时间
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timerService().currentProcessingTime();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; // 获取当前事件时间
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timerService().currentWatermark();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; // 注册系统处理时间定时器
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timerService().registerProcessingTimeTimer(& nbsp; ctx.timerService().currentProcessingTime()& nbsp; +& nbsp; 1000L);
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; tsTimeState.update(& nbsp; ctx.timerService().currentProcessingTime()& nbsp; +& nbsp; 1000L);

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; // 注册事件时间定时器
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //ctx.timerService().registerEventTimeTimer((value.getTimestamp() + 10) * 1000L);
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; // 删除时间
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //ctx.timerService().deleteProcessingTimeTimer(tsTimeState.value());
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; @Override
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; void& nbsp; onTimer(long& nbsp; timestamp,& nbsp; OnTimerContext& nbsp; ctx,& nbsp; Collector& lt; Integer& gt; & nbsp; out)& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; System.out.println(timestamp+& quot; 定时器触发& quot; );

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.getCurrentKey();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; //ctx.output();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; ctx.timeDomain();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp;

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; @Override
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; void& nbsp; close()& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; tsTimeState.clear();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp;
& nbsp; & nbsp; & nbsp; & nbsp;

一个案例：

监控温度传感器的温度值，如果温度值在 10 秒钟之内连续上升，则报警。

public& nbsp; class& nbsp; ProcessTest2_ApplicationCase& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; public& nbsp; static& nbsp; void& nbsp; main(String[]& nbsp; args)& nbsp; throws& nbsp; Exception& nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; StreamExecutionEnvironment& nbsp; env& nbsp; =& nbsp; StreamExecutionEnvironment.getExecutionEnvironment();
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; env.setParallelism(1);

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; DataStreamSource& lt; String& gt; & nbsp; inputStream& nbsp; =& nbsp; env.socketTextStream(& quot; localhost& quot; ,& nbsp; 7777);
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; DataStream& lt; SensorReading& gt; & nbsp; dataStream& nbsp; =& nbsp; inputStream.map(line& nbsp; -& gt; & nbsp;
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; String[]& nbsp; fields& nbsp; =& nbsp; line.split(& quot; ,& quot; );
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; return& nbsp; new& nbsp; SensorReading(fields[0],& nbsp; new& nbsp; Long(fields[1]),& nbsp; new& nbsp; Double(fields[2]));
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; );

& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; dataStream.keyBy(& quot; id& quot; )
& nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; & nbsp; .process(

推荐阅读

如何激活电子医保卡? 如何激活mongodb

空气炸锅烤橘子温度和时间空气炸锅烤橘子要多久

《冰与火之歌》《权力的游戏》第八季,你觉得最终谁会坐上七国的铁王座？

ps流线分析图,ps怎么做流线分析箭头

关于739游戏棋牌的信息

软件实用性分析报告,中国软件股票分析报告

功能机|售价249元，诺基亚发布新款功能机，外形方正能扫码

亚马逊|亚马逊封店潮背后，溃散的实体工厂与芯片炒货客

新疆最好的葡萄品种是什么品种中国最好吃的葡萄品种排名

黄码怎么变绿码需要几次核酸检测

K3有没有上市? k3为什么停产

北京生育津贴领取条件及标准2023最新?北京生育津贴怎么计算?

用Photoshop给穿古装的云浮女孩调些暗青色效果

中移电子华院是什么中移电商华院是什么平台，中移电子商务有限公司-华院

吃海鲜可以喝茶吗

资生堂哪几款产品值得购买？

税务季度申报如何网上申报企业所得税哪里申报，一般纳税人企业所得税

想创业,有没有好点的冷门项目？

一日为师，终身为父

临安房屋出租信息最新临安房屋出租信息

bat脚本系列---批量修改文件名

jenkins系列---jenkins+gitee+vue持续集成到另一台服务器

linux之同时监控多个日志文件变化

elk

web服务Tomcat优化之多实例

更改AD DC的电脑名称

Mac 终端窗口的美化

CentOS安装zip unzip命令

CS162操作系统课程第二课-4个核心OS概念