Flink中的ProcessFunction API

浅浅的花香味﹌ 2023-01-17 09:16 112阅读 0赞

## 一、背景介绍 ##

之前学习的转换算子是无法访问事件的时间戳信息和水位线信息的，而这些信息在一些应用场景是比较重要的。例如，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（窗口处理）

## 二、示例 ##

### 2.1 ###

### 2.2 侧输出流(SideOutput) ###

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

public class ProcessTest2_SideOutputCase { 
    
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            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();
        }
    }