4.1.20 Flink-流处理框架-ProcessFunction API(底层 API)

2024-02-28 05:18

本文主要是介绍4.1.20 Flink-流处理框架-ProcessFunction API(底层 API),希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

目录

1.写在前面

2.Process Function中的细节

2.1 KeyedProcessFunction(重点介绍)

2.1.1 TimerService 和 定时器(Timers)

2.1.2 基本代码演示

2.1.3 需求实例

2.2 侧输出流(SideOutput)

2.3 CoProcessFunction


1.写在前面

        我们之前了解过,flink有分层API, 最底层级的抽象仅仅提供了有状态流,它将通过过程函数(Process Function) 被嵌入到 DataStream API 中。底层过程函数(Process Function) 与 DataStream API 相集成,使其可以对某些特定的操作进行底层的抽象,它允许用户可以自由地处理来自一个或多个数据流的事件,并使用一致的容错的状态。除此之外,用户可以注册事件时间并处理时间回调,从而使程序可以处理复杂的计算。

        我们之前学习的转换算子是无法访问事件的时间戳信息和水位线信息的。而这在一些应用场景下,极为重要。例如 MapFunction 这样的 map 转换算子就无法访问 时间戳或者当前事件的事件时间。基于此,DataStream API 提供了一系列的 Low-Level 转换算子。可以访问时间戳、watermark 以及注册定时事件(一段时间之后执行某个方法)。还可以输出特定的一些事件,例如超时事件等。 Process Function 用来构建事件驱动的应用以及实现自定义的业务逻辑(使用之前的 window 函数和转换算子无法实现)。例如,Flink SQL 就是使用 Process Function 实现的

        Flink 提供了 8 个 Process Function:(1)ProcessFunction (2)KeyedProcessFunction(3)CoProcessFunction(4)ProcessJoinFunction(5)BroadcastProcessFunction(6)KeyedBroadcastProcessFunction(7)ProcessWindowFunction(8)ProcessAllWindowFunction

2.Process Function中的细节

2.1 KeyedProcessFunction(重点介绍)

        这里我们重点介绍 KeyedProcessFunction。KeyedProcessFunction 用来操作 KeyedStream。KeyedProcessFunction 会处理流 的每一个元素,输出为 0 个、1 个或者多个元素。所有的 Process Function 都继承自 RichFunction 接口,所以都有 open()、close()和 getRuntimeContext()等方法。而 KeyedProcessFunction还额外提供了两个方法:

  •         (1)processElement(I value, Context ctx, Collector out), 输入流中的每一个元素都会调用这个方法,调用结果将会放在 Collector 数据类型中输出。Context 可以访问元素的时间戳,元素的 key,以及 TimerService 时间服务。Context 还可以将结果输出到别的流(side outputs)。
  •         (2)onTimer(long timestamp, OnTimerContext ctx, Collector out) 是一个回调函数。当之前注册的定时器触发时调用。参数 timestamp 为定时器所设定的触发的时间戳。Collector 为输出结果的集合。OnTimerContext 和 processElement 的 Context 参数一样,提供了上下文的一些信息,例如定时器触发的时间信息(事件时间或者处理时间)

2.1.1 TimerService 和 定时器(Timers)

        Context 和 OnTimerContext 所持有的 TimerService 对象拥有以下方法:

  1. • long currentProcessingTime() 返回当前处理时间
  2. • long currentWatermark() 返回当前 watermark 的时间戳
  3. • void registerProcessingTimeTimer(long timestamp) 会注册当前 key 的 processing time 的定时器。当 processing time 到达定时时间时,触发 timer。
  4. • void registerEventTimeTimer(long timestamp) 会注册当前 key 的 event time 定时器。当水位线大于等于定时器注册的时间时,触发定时器执行回调函数。
  5. • void deleteProcessingTimeTimer(long timestamp) 删除之前注册处理时间定时器。如果没有这个时间戳的定时器,则不执行。
  6. • void deleteEventTimeTimer(long timestamp) 删除之前注册的事件时间定时器,如果没有此时间戳的定时器,则不执行。

        当定时器 timer 触发时,会执行回调函数 onTimer()。注意定时器 timer 只能在 keyed streams 上面使用。

2.1.2 基本代码演示

        举个例子说明 KeyedProcessFunction 如何操作 KeyedStream。

package com.ucas.apitest.processfunction;import com.atguigu.apitest.beans.SensorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;public class ProcessTest1_KeyedProcessFunction {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// socket文本流DataStream<String> inputStream = env.socketTextStream("localhost", 7777);// 转换成SensorReading类型DataStream<SensorReading> dataStream = inputStream.map(line -> {String[] fields = line.split(",");return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));});// 测试KeyedProcessFunction,先分组然后自定义处理dataStream.keyBy("id").process( new MyProcess() ).print();env.execute();}// 实现自定义的处理函数public static class MyProcess extends KeyedProcessFunction<Tuple, SensorReading, Integer>{ValueState<Long> tsTimerState;@Overridepublic void open(Configuration parameters) throws Exception {tsTimerState =  getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts-timer", Long.class));}@Overridepublic void processElement(SensorReading value, Context ctx, Collector<Integer> out) throws Exception {out.collect(value.getId().length());// contextctx.timestamp();ctx.getCurrentKey();
//            ctx.output();ctx.timerService().currentProcessingTime();ctx.timerService().currentWatermark();ctx.timerService().registerProcessingTimeTimer( ctx.timerService().currentProcessingTime() + 5000L);tsTimerState.update(ctx.timerService().currentProcessingTime() + 1000L);
//            ctx.timerService().registerEventTimeTimer((value.getTimestamp() + 10) * 1000L);
//            ctx.timerService().deleteProcessingTimeTimer(tsTimerState.value());}@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {System.out.println(timestamp + " 定时器触发");ctx.getCurrentKey();
//            ctx.output();ctx.timeDomain();}@Overridepublic void close() throws Exception {tsTimerState.clear();}}
}

2.1.3 需求实例

        需求:监控温度传感器的温度值,如果温度值在 10 秒钟之内(processing time) 连续上升,则报警。

        看一下 TempIncreaseWarning 如何实现, 程序中使用 ValueState 状态变量来保存上次的温度值和定时器时间戳

代码演示:

package com.atguigu.apitest.processfunction;
import com.atguigu.apitest.beans.SensorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;public class ProcessTest2_ApplicationCase {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// socket文本流DataStream<String> inputStream = env.socketTextStream("localhost", 7777);// 转换成SensorReading类型DataStream<SensorReading> dataStream = inputStream.map(line -> {String[] fields = line.split(",");return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));});// 测试KeyedProcessFunction,先分组然后自定义处理dataStream.keyBy("id").process( new TempConsIncreWarning(10) ).print();env.execute();}// 实现自定义处理函数,检测一段时间内的温度连续上升,输出报警public static class TempConsIncreWarning extends KeyedProcessFunction<Tuple, SensorReading, String>{// 定义私有属性,当前统计的时间间隔private Integer interval;public TempConsIncreWarning(Integer interval) {this.interval = interval;}// 定义状态,保存上一次的温度值,定时器时间戳private ValueState<Double> lastTempState;private ValueState<Long> timerTsState;@Overridepublic void open(Configuration parameters) throws Exception {lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp", Double.class, Double.MIN_VALUE));timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts", Long.class));}@Overridepublic void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {// 取出状态Double lastTemp = lastTempState.value();Long timerTs = timerTsState.value();// 如果温度上升并且没有定时器,注册10秒后的定时器,开始等待if( value.getTemperature() > lastTemp && timerTs == null ){// 计算出定时器时间戳Long ts = ctx.timerService().currentProcessingTime() + interval * 1000L;ctx.timerService().registerProcessingTimeTimer(ts);timerTsState.update(ts);}// 如果温度下降,那么删除定时器else if( value.getTemperature() < lastTemp && timerTs != null ){ctx.timerService().deleteProcessingTimeTimer(timerTs);timerTsState.clear();}// 更新温度状态lastTempState.update(value.getTemperature());}@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {// 定时器触发,输出报警信息out.collect("传感器" + ctx.getCurrentKey().getField(0) + "温度值连续" + interval + "s上升");timerTsState.clear();}@Overridepublic void close() throws Exception {lastTempState.clear();}}
}

2.2 侧输出流(SideOutput)

        大部分的 DataStream API 的算子的输出是单一输出,也就是某种数据类型的流。 除了 split 算子,可以将一条流分成多条流,这些流的数据类型也都相同。process function 的 side outputs 功能可以产生多条流,并且这些流的数据类型可以不一样。 一个 side output 可以定义为 OutputTag[X]对象,X 是输出流的数据类型。process function 可以通过 Context 对象发射一个事件到一个或者多个 side outputs

        下面是一个示例程序,用来监控传感器温度值,将温度值低于 30 度的数据输出到 side output

package com.atguigu.apitest.processfunction;
import com.atguigu.apitest.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;public class ProcessTest3_SideOuptCase {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// socket文本流DataStream<String> inputStream = env.socketTextStream("localhost", 7777);// 转换成SensorReading类型DataStream<SensorReading> dataStream = inputStream.map(line -> {String[] fields = line.split(",");return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));});// 定义一个OutputTag,用来表示侧输出流低温流OutputTag<SensorReading> lowTempTag = new OutputTag<SensorReading>("lowTemp") {};// 测试ProcessFunction,自定义侧输出流实现分流操作SingleOutputStreamOperator<SensorReading> highTempStream = dataStream.process(new ProcessFunction<SensorReading, SensorReading>() {@Overridepublic void processElement(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {// 判断温度,大于30度,高温流输出到主流;小于低温流输出到侧输出流if( value.getTemperature() > 30 ){out.collect(value);}else {ctx.output(lowTempTag, value);}}});highTempStream.print("high-temp");highTempStream.getSideOutput(lowTempTag).print("low-temp");env.execute();}
}

2.3 CoProcessFunction

        对于两条输入流,DataStream API 提供了 CoProcessFunction 这样的 low-level 操作。CoProcessFunction 提供了操作每一个输入流的方法: processElement1()和 processElement2()。

        类似于 ProcessFunction,这两种方法都通过 Context 对象来调用。这个 Context 对象可以访问事件数据,定时器时间戳,TimerService,以及 side outputs。 CoProcessFunction 也提供了 onTimer()回调函数。

这篇关于4.1.20 Flink-流处理框架-ProcessFunction API(底层 API)的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



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

相关文章

GSON框架下将百度天气JSON数据转JavaBean

《GSON框架下将百度天气JSON数据转JavaBean》这篇文章主要为大家详细介绍了如何在GSON框架下实现将百度天气JSON数据转JavaBean,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下... 目录前言一、百度天气jsON1、请求参数2、返回参数3、属性映射二、GSON属性映射实战1、类对象映

Python自动化处理PDF文档的操作完整指南

《Python自动化处理PDF文档的操作完整指南》在办公自动化中,PDF文档处理是一项常见需求,本文将介绍如何使用Python实现PDF文档的自动化处理,感兴趣的小伙伴可以跟随小编一起学习一下... 目录使用pymupdf读写PDF文件基本概念安装pymupdf提取文本内容提取图像添加水印使用pdfplum

C# LiteDB处理时间序列数据的高性能解决方案

《C#LiteDB处理时间序列数据的高性能解决方案》LiteDB作为.NET生态下的轻量级嵌入式NoSQL数据库,一直是时间序列处理的优选方案,本文将为大家大家简单介绍一下LiteDB处理时间序列数... 目录为什么选择LiteDB处理时间序列数据第一章:LiteDB时间序列数据模型设计1.1 核心设计原则

基于Redis自动过期的流处理暂停机制

《基于Redis自动过期的流处理暂停机制》基于Redis自动过期的流处理暂停机制是一种高效、可靠且易于实现的解决方案,防止延时过大的数据影响实时处理自动恢复处理,以避免积压的数据影响实时性,下面就来详... 目录核心思路代码实现1. 初始化Redis连接和键前缀2. 接收数据时检查暂停状态3. 检测到延时过

Java利用@SneakyThrows注解提升异常处理效率详解

《Java利用@SneakyThrows注解提升异常处理效率详解》这篇文章将深度剖析@SneakyThrows的原理,用法,适用场景以及隐藏的陷阱,看看它如何让Java异常处理效率飙升50%,感兴趣的... 目录前言一、检查型异常的“诅咒”:为什么Java开发者讨厌它1.1 检查型异常的痛点1.2 为什么说

Python利用PySpark和Kafka实现流处理引擎构建指南

《Python利用PySpark和Kafka实现流处理引擎构建指南》本文将深入解剖基于Python的实时处理黄金组合:Kafka(分布式消息队列)与PySpark(分布式计算引擎)的化学反应,并构建一... 目录引言:数据洪流时代的生存法则第一章 Kafka:数据世界的中央神经系统消息引擎核心设计哲学高吞吐

C++ STL-string类底层实现过程

《C++STL-string类底层实现过程》本文实现了一个简易的string类,涵盖动态数组存储、深拷贝机制、迭代器支持、容量调整、字符串修改、运算符重载等功能,模拟标准string核心特性,重点强... 目录实现框架一、默认成员函数1.默认构造函数2.构造函数3.拷贝构造函数(重点)4.赋值运算符重载函数

Go语言使用Gin处理路由参数和查询参数

《Go语言使用Gin处理路由参数和查询参数》在WebAPI开发中,处理路由参数(PathParameter)和查询参数(QueryParameter)是非常常见的需求,下面我们就来看看Go语言... 目录一、路由参数 vs 查询参数二、Gin 获取路由参数和查询参数三、示例代码四、运行与测试1. 测试编程路

Java异常捕获及处理方式详解

《Java异常捕获及处理方式详解》异常处理是Java编程中非常重要的一部分,它允许我们在程序运行时捕获并处理错误或不预期的行为,而不是让程序直接崩溃,本文将介绍Java中如何捕获异常,以及常用的异常处... 目录前言什么是异常?Java异常的基本语法解释:1. 捕获异常并处理示例1:捕获并处理单个异常解释:

Go语言使用net/http构建一个RESTful API的示例代码

《Go语言使用net/http构建一个RESTfulAPI的示例代码》Go的标准库net/http提供了构建Web服务所需的强大功能,虽然众多第三方框架(如Gin、Echo)已经封装了很多功能,但... 目录引言一、什么是 RESTful API?二、实战目标:用户信息管理 API三、代码实现1. 用户数据