Flink ProcessFunction 使用总结

什么是ProcessFunction

ProcessFunction是一个低层次的流处理操作，允许返回所有(无环的)流程序的基础构建模块:

事件(event)(流元素)状态(state)(容错性，一致性，仅在keyed stream中)定时器(timers)(event time和processing time，仅在keyed stream中)

ProcessFunction可以认为是能够访问到keyed state和timers的FlatMapFunction，输入流中接收到的每个事件都会调用它来处理。

对于容错性状态，ProcessFunction可以通过RuntimeContext来访问Flink的keyed state，方法与其他状态性函数访问keyed state一样。

定时器允许应用程序对processing time和event time的变化做出反应，每次对processElement(...)的调用都会得到一个Context对象，该对象允许访问元素事件时间的时间戳和TimeServer。TimeServer可以用来为尚未发生的event-time或者processing-time注册回调，当定时器的时间到达时，onTimer(...)方法会被调用。在这个调用期间，所有的状态都会限定到创建定时器的键，并允许定时器操纵键控状态(keyed states)。

为什么使用ProcessFunction

在之前学习得转换算子，是无法访问事件得时间戳信息以及水位线信息得。而这个在一些场景下，又是极为重要得。比如说，MapFunction这样的map转换算子就无法访问时间戳或者当前事件的事件时间；

基于此，DataStream API 提供了一些列的 Low-Level转换算子，可以访问时间戳，watermark以及注册定时事件。还可以输出特定的一些事件，比如超时事件等。

ProcessFunction 用来构建事件驱动的应用以及实现自定义的业务逻辑（使用之前的window函数和转换算子无法实现），例如，Flink SQL就是使用 Process Function 实现的；

Flink提供了8个ProcessFunction：

ProcessFunctionKeyedProcessFunctionCoProcessFunctionProcessJoinFunctionBroadcastProcessFunctionKeyedBroadcastProcessFunctionProcessWindowFunctionProcessAllWindowFunction

下面以ProcessFunction为例，看看其简单的API使用代码实现的具体案例

import com.congge.source.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 KeyedProcessFunction1 {

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

        // socket文本流
        DataStream inputStream = env.socketTextStream("localhost", 7777);

        // 转换成SensorReading类型
        DataStream 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 {

        ValueState tsTimerState;

        @Override
        public void open(Configuration parameters) throws Exception {
            tsTimerState = getRuntimeContext().getState(new ValueStateDescriptor("ts-timer", Long.class));
        }

        @Override
        public void processElement(SensorReading value, Context ctx, Collector 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 out) throws Exception {
            System.out.println(timestamp + " 定时器触发");
            ctx.getCurrentKey();
//            ctx.output();
            ctx.timeDomain();
        }

        @Override
        public void close() throws Exception {
            tsTimerState.clear();
        }
    }

}

场景应用

有这样一个场景，业务上要求在一个可预测的时间范围内，如果输入的数据流中，这段时间内的某个业务字段的数值一直诚信递增或者递减的趋势，程序则给出相应的告警；

这是一个很好的可以使用ProcessFunction 实现的场景，下面来看具体的代码实现，

import com.congge.source.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 ApplicationCase1 {

    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // socket文本流
        DataStream inputStream = env.socketTextStream("localhost", 7777);

        // 转换成SensorReading类型
        DataStream 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{

        // 定义私有属性，当前统计的时间间隔
        private Integer interval;

        public TempConsIncreWarning(Integer interval) {
            this.interval = interval;
        }

        // 定义状态，保存上一次的温度值，定时器时间戳
        private ValueState lastTempState;

        private ValueState timerTsState;

        @Override
        public void open(Configuration parameters) throws Exception {
            lastTempState = getRuntimeContext().getState(new ValueStateDescriptor("last-temp", Double.class, Double.MIN_VALUE));
            timerTsState = getRuntimeContext().getState(new ValueStateDescriptor("timer-ts", Long.class));
        }

        @Override
        public void processElement(SensorReading value, Context ctx, Collector 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 out) throws Exception {
            // 定时器触发，输出报警信息
            out.collect("传感器" + ctx.getCurrentKey().getField(0) + "温度值连续" + interval + "s上升");
            timerTsState.clear();
        }

        @Override
        public void close() throws Exception {
            lastTempState.clear();
        }
    }

}

Flink ProcessFunction 使用总结

大数据系统相关栏目本月热门文章