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

2022 年 7 月 04 日
本文字数：4708 字
阅读完需：约 15 分钟

概述

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

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 value, Context ctx, Collector<O> out)

复制代码

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

onTimer(long timestamp, OnTimerContext ctx, Collector<O> out)

复制代码

定时器

Context 中 TimerService 对象方汇总：

返回当前处理时间

long currentProcessingTime()

复制代码

返回当前 watermark 的时间戳

long currentWatermark()

复制代码

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

void registerProcessingTimeTimer(long timestamp)

复制代码

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

void registerEventTimeTimer(long timestamp)

复制代码

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

void deleteProcessingTimeTimer(long timestamp)

复制代码

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

void deleteEventTimeTimer(long timestamp)

复制代码

测试代码：

public class ProcessTest1_KeyedProcessFunction {    public static void main(String[] args) throws Exception{        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setParallelism(1);
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 7777);        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> tsTimeState;
        @Override        public void open(Configuration parameters) throws Exception {            tsTimeState = getRuntimeContext().getState(new ValueStateDescriptor<Long>(                    "tsTimeState",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() + 1000L);            tsTimeState.update( ctx.timerService().currentProcessingTime() + 1000L);
            // 注册事件时间定时器            //ctx.timerService().registerEventTimeTimer((value.getTimestamp() + 10) * 1000L);            // 删除时间            //ctx.timerService().deleteProcessingTimeTimer(tsTimeState.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 {            tsTimeState.clear();        }    }}

复制代码

一个案例：

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

public class ProcessTest2_ApplicationCase {    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setParallelism(1);
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 7777);        DataStream<SensorReading> dataStream = inputStream.map(line -> {            String[] fields = line.split(",");            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));        });

        dataStream.keyBy("id")                .process(new TempConsIncreWarring(10))                .print();
        env.execute();    }
    // 自定义函数 检测一段时间（时间域）内温度连续上升，输出报警    private static class TempConsIncreWarring extends KeyedProcessFunction<Tuple, SensorReading, String> {        // 定义时间域        private Integer interval;
        public TempConsIncreWarring(Integer interval) {            this.interval = interval;        }
        // 定义状态，保存上一个温度值，定时器时间戳        private ValueState<Double> lastTempState;        private ValueState<Long> tsTimeState;
        @Override        public void open(Configuration parameters) throws Exception {            lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>(                    "lastTempState", Double.class, Double.MIN_VALUE));            tsTimeState = getRuntimeContext().getState(new ValueStateDescriptor<Long>(                    "tsTimeState", Long.class));        }
        @Override        public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {            //  取出状态            Double lastTemp = lastTempState.value();            Long tsTime = tsTimeState.value();
            //  如果温度上升就要注册10秒后的定时器且在无定时器时，等待            if (value.getTemperature() > lastTemp && tsTime == null) {                // 计算出定时器时间戳                Long ts = ctx.timerService().currentProcessingTime() + interval * 1000L;                // 注册定时器                ctx.timerService().registerProcessingTimeTimer(ts);                // 更新时间状态                tsTimeState.update(ts);            }            // 如果温度下降 需要删除定时器            if (value.getTemperature() < lastTemp && tsTime != null) {                ctx.timerService().deleteProcessingTimeTimer(tsTime);                // 清空时间状态定时器                tsTimeState.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 +"秒一直处于上升");            tsTimeState.clear();        }
        @Override        public void close() throws Exception {            lastTempState.clear();        }    }}

复制代码

运行结果：

侧流输出

侧流输出功能可以产生多条流，并且这些流的数据类型可以不一样。一个侧流可以定义为 OutputTag[X]对象，X 是输出流的数据类型。

一个案例：

监控传感器温度值，将温度值低于 30 度的数据输出到 SideOutput

public class ProcessTest3_SideOutputCase {    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setParallelism(1);
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 7777);
        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 lowTemp = new OutputTag<SensorReading>("lowTemp") {        };
        // 自定义测输出流实现分流操作        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(lowTemp, value);                }            }        });
        highTempStream.print("high-Temp");        highTempStream.getSideOutput(lowTemp).print("low");

        env.execute();    }}