Flink：Process Function

ProcessFunction API（底层 API）

DataStream API 提供了一系列的 Low-Level 转换算子。可以访问时间戳、watermark 以及注册定时事件。还可以输出特定的一些事件，例如超时事件等。Process Function 用来构建事件驱动的应用以及实现自定义的业务逻辑(使用之前的 window 函数和转换算子无法实现)。例如，Flink SQL 就是使用 Process Function 实现的。

Flink提供了8个Process Function：

ProcessFunction
KeyedProcessFunction
CoProcessFunction
ProcessJoinFunction
BroadcastProcessFunction
KeyedBroadcastProcessFunction
ProcessWindowFunction
ProcessAllWindowFunction

KeyedProcessFunction

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

processElement(I value, Context ctx, Collector out):流中的每一个元素都会调用这个方法，调用结果将会放在 Collector 数据类型中输出。Context 可以访问元素的时间戳，元素的 key，以及 TimerService 时间服务。Context 还可以将结果输出到别的流(side outputs)。

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

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 上面使用。

测试：

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

        DataStream<String> inputStream = env.readTextFile("D:\project\flink-demo\src\main\resources\sensor.txt");
        DataStream<SensorReading> mapStream = inputStream.map((str) -> {
            String[] split = str.split(" ");
            return new SensorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2]));
        });

        mapStream.keyBy("id")
                .process(new MyProcess()).print();

        env.execute();


    }

    public static class MyProcess extends KeyedProcessFunction<Tuple, SensorReading, Integer> {

        private ValueState<Long> timer;

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

        @Override
        public void processElement(SensorReading value, Context ctx, Collector<Integer> out) throws Exception {
            out.collect(value.getId().length());

            //获取当前key
            Tuple currentKey = ctx.getCurrentKey();
            Object field = currentKey.getField(0);
            System.out.println(field);


//            System.out.println(ctx.timerService().currentProcessingTime());
//            System.out.println(ctx.timerService().currentWatermark());
            System.out.println(ctx.timerService().currentProcessingTime());
            ctx.timerService().registerProcessingTimeTimer( ctx.timerService().currentProcessingTime() + 500L);
            timer.update(ctx.timerService().currentProcessingTime() + 1000);
            //注册
            //ctx.timerService().deleteProcessingTimeTimer(timer.value());
            TimeUnit.SECONDS.sleep(1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
            System.out.println("触发定时器onTimer：" + timestamp);
        }
    }

效果：

应用案例

一段时间内温度连续上升

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

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

        DataStream<String> inputStream = env.socketTextStream("192.168.1.77", 7777);
        DataStream<SensorReading> mapStream = inputStream.map((str) -> {
            String[] split = str.split(" ");
            return new SensorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2]));
        });

        mapStream.keyBy("id")
                .process(new WarningFunction(6)).print();

        env.execute();
    }

    public static class WarningFunction extends KeyedProcessFunction<Tuple, SensorReading, String> {

        private Integer interval;

        //定义状态,保存上一次的温度值和定时器时间戳
        private ValueState<Double> lastTempState;
        private ValueState<Long> timerState;

        @Override
        public void open(Configuration parameters) throws Exception {
            lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<>("lastTempState", Double.class, Double.MIN_VALUE));
            timerState = getRuntimeContext().getState(new ValueStateDescriptor<>("timerState", Long.class));
        }

        @Override
        public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
            Double lastTemp = lastTempState.value();
            Long timerTs = timerState.value();

            //温度是否上升
            if (value.getTemperature() > lastTemp && timerTs == null){
                //计算出定时器时间戳
                long ts = ctx.timerService().currentProcessingTime() + interval * 1000L;
                ctx.timerService().registerProcessingTimeTimer(ts);
                timerState.update(ts);
            }else if (value.getTemperature() < lastTemp && timerTs != null){
                ctx.timerService().deleteProcessingTimeTimer(timerTs);
                timerState.clear();
            }

            //更新温度状态
            lastTempState.update(value.getTemperature());

        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            out.collect("传感器"+ctx.getCurrentKey().getField(0)+"温度连续上升");
            timerState.clear();
        }

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

测试效果：

测输出流

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

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

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

        DataStream<String> inputStream = env.readTextFile("D:\project\flink-demo\src\main\resources\sensor.txt");
        DataStream<SensorReading> mapStream = inputStream.map((str) -> {
            String[] split = str.split(" ");
            return new SensorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2]));
        });

        OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("low"){};

        SingleOutputStreamOperator<SensorReading> highStream = mapStream.process(new ProcessFunction<SensorReading, SensorReading>() {
            @Override
            public void processElement(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {
                if (value.getTemperature() > 30){
                    out.collect(value);
                }else {
                    ctx.output(outputTag, value);
                }
            }
        });
        DataStream<SensorReading> low = highStream.getSideOutput(outputTag);
        low.print("low");
        highStream.print("high");
        env.execute();
    }

sensor.txt

sensor_1 1547718199 35.8
sensor_2 1547718199 16.8
sensor_3 1547718199 26.9
sensor_1 1547718199 17.8
sensor_2 1547718199 38.8
sensor_3 1547718199 39.8

测试输出：