4.1.20 Flink-流处理框架-ProcessFunction API（底层 API）

本文主要是介绍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 对象拥有以下方法:

• long currentProcessingTime() 返回当前处理时间
• long currentWatermark() 返回当前 watermark 的时间戳
• void registerProcessingTimeTimer(long timestamp) 会注册当前 key 的 processing time 的定时器。当 processing time 到达定时时间时，触发 timer。
• void registerEventTimeTimer(long timestamp) 会注册当前 key 的 event time 定时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。
• void deleteProcessingTimeTimer(long timestamp) 删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行。
• 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();}
}