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

痛定思痛。 2022-11-07 01:57 168阅读 0赞

**目录**

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 相集成，使其可以对某些特定的操作进行底层的抽象，它允许用户可以自由地处理来自一个或多个数据流的事件，并使用一致的容错的状态。除此之外，用户可以注册事件时间并处理时间回调，从而使程序可以处理复杂的计算。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

我们之前学习的转换算子是无法访问事件的时间戳信息和水位线信息的。而这在一些应用场景下，极为重要。例如 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 参数一样，提供了上下文的一些信息，例如定时器触发的时间信息(事件时间或者处理时间)**。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

### 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;
    
            @Override
            public void open(Configuration parameters) throws Exception {
                tsTimerState =  getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts-timer", Long.class));
            }
    
            @Override
            public void processElement(SensorReading value, Context ctx, Collector<Integer> out) throws Exception {
                out.collect(value.getId().length());
    
                // context
                ctx.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());
            }
    
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
                System.out.println(timestamp + " 定时器触发");
                ctx.getCurrentKey();
    //            ctx.output();
                ctx.timeDomain();
            }
    
            @Override
            public void close() throws Exception {
                tsTimerState.clear();
            }
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

### 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;
    
            @Override
            public 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));
            }
    
            @Override
            public 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());
            }
    
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                // 定时器触发，输出报警信息
                out.collect("传感器" + ctx.getCurrentKey().getField(0) + "温度值连续" + interval + "s上升");
                timerTsState.clear();
            }
    
            @Override
            public 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>() {
                @Override
                public 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()回调函数。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/721de6eb08ff4d1bb9eb4110d3b6ea7c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221023/b507e9a81b8647d39ab2edc008ce12f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221023/e17ec9cad38f43ed9290e18885ca1257.png