Flink-底层API（ProcessFunction API）学习使用

客官°小女子只卖身不卖艺 2022-09-09 10:58 116阅读 0赞

# ProcessFunction API(底层 API) #

**转换算子** 是无法访问事件的时间戳信息和 水位线信息的。而这在一些应用场景下,极为重要。例如 MapFunction 这样的 map 转换算子就无法访问时间戳或者当前事件的事件时间。

基于此,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。

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

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 参数一样,提供了上下文的一些信息,例如定时器触发的时间信息(事件时间或者处理时间 )。

简单KeyedProcessFunction使用

public static void main(String[] args) throws Exception { 
    
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
    
        // 从Socket读取
        DataStream<String> inputStream = env.socketTextStream("localhost", 4444);
    
        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 MyKeyedProcess()).print();
    
    
        // 执行
        env.execute();
    }
    
    // 实现自定义的处理函数
    public static class MyKeyedProcess extends KeyedProcessFunction<Tuple, SensorReading, Integer> { 
    
        private ValueState<Long> tsTimer;
    
        @Override
        public void open(Configuration parameters) throws Exception { 
    
            tsTimer = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts-timer", Long.class));
        }
    
        @Override
        public void processElement(SensorReading value, Context context, Collector<Integer> out) throws Exception { 
    
            out.collect(value.getId().length());
    
    
            // context
            context.timestamp(); // 数据的时间戳
            context.getCurrentKey(); // 当前的key
            // context.output(); // 定义侧输出流
            TimerService timerService = context.timerService();// 获取定时服务
            timerService.currentProcessingTime();  // 当前处理时间
            timerService.currentWatermark();       // 当前事件时间，WM
            // 注册处理时间的定时器
            long interval = timerService.currentProcessingTime() + 3000;
            timerService.registerProcessingTimeTimer(interval);
            tsTimer.update(interval);
            // 注册事件时间的定时器
            // timerService.registerEventTimeTimer((value.getTimestamp() + 10) * 1000);
            // 删除处理定时器
            // timerService.deleteProcessingTimeTimer(tsTimer.value());
            // 删除事件定时器
            // timerService.deleteEventTimeTimer((value.getTimestamp() + 10) * 1000);
        }
    
        // 定时器执行具体的业务
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception { 
    
            System.out.println(timestamp + "定时器触发");
    
            ctx.getCurrentKey();
            ctx.timeDomain();
        }
    
        @Override
        public void close() throws Exception { 
            tsTimer.clear();
        }
    }
    
    lxj@lxj:~$ nc -lk 4444
    sensor_1,1547718209,31
    
    8
    1630398317849定时器触发

## 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 只 能在keyedstreams上面使用 。

**KeyedProcessFunction 操作 KeyedStream例子**：需求:监控温度传感器的温度值,如果温度值在 10 秒钟之内(processing time)连续上升,则报警。

public static void main(String[] args) throws Exception { 
    
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
    
        // 从Socket读取
        DataStream<String> inputStream = env.socketTextStream("localhost", 4444);
    
        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 MyKeyedProcessCase(10)).print();
    
    
        // 执行
        env.execute();
    }
    
    // 实现自定义的处理函数
    public static class MyKeyedProcessCase extends KeyedProcessFunction<Tuple, SensorReading, String> { 
    
        private Integer interval;
        // 定义状态，保存上一次的温度值，定时器时间戳
        private ValueState<Double> lastTempState;
        private ValueState<Long> timerTsState;
    
        public MyKeyedProcessCase(Integer interval) { 
            this.interval = interval;
        }
    
        @Override
        public void open(Configuration parameters) throws Exception { 
    
            lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-timp", Double.class));
            lastTempState.update(Double.MAX_VALUE);
    
            timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts", Long.class));
        }
    
        @Override
        public void processElement(SensorReading value, Context context, Collector<String> out) throws Exception { 
    
            // 取出状态
            Double lastTemp = lastTempState.value();
            Long timerTs = timerTsState.value();
    
            // 如果温度上升并且没有定时器，注册10s后的定时器，并开始等待
            if (value.getTemperature() > lastTemp && timerTs == null) { 
                // 计算出定时器时间戳
                long ts = context.timerService().currentProcessingTime() + interval * 1000;
                context.timerService().registerProcessingTimeTimer(ts);
    
                timerTsState.update(ts);
            } else if (value.getTemperature() < lastTemp && timerTs != null) { 
                // 如果温度下降并且有定时器，那么删除定时器
                context.timerService().deleteProcessingTimeTimer(timerTsState.value());
    
                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();
            timerTsState.clear();
        }
    }

## 侧输出流(SideOutput) ##

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

需求：监控传感器温度值,将温度值低于 30 度的数据输出到 side output。

## CoProcessFunction ##

对于两条输入流,DataStream API 提供了 CoProcessFunction 这样的 low-level操作。CoProcessFunction 提供了操作每一个输入流的方法: processElement1()和processElement2()。

类似于 ProcessFunction,这两种方法都通过 Context 对象来调用。这个 Context对象可以访问事件数据,定时器时间戳,TimerService,以及 side outputs。CoProcessFunction 也提供了 onTimer()回调函数。