• Flink的处理函数——processFunction

    目录

    一、处理函数概述       

    二、Process函数分类——8个

    （1）ProcessFunction

    （2）KeyedProcessFunction

    （3）ProcessWindowFunction

    （4）ProcessAllWindowFunction

    （5）CoProcessFunction

    （6）ProcessJoinFunction

    （7）BroadcastProcessFunction

    （8）KeyedBroadcastProcessFunction

    三、KeyedProcessFunction案例

    1.运行processElement方法中的事件时间

    (1)输入数据

    2.运行processElement方法中的处理时间

    (1)先输入一条数据

    (2)再快速输入两条数据

    3.运行processElement方法中的水位线

    (1)输入数据

    4.总结：

    四、应用案例——求TopN

    (一)思路一：hashmap

    (二)思路二：keyby

    (三)思路三：使用侧输出流——推荐

    一、处理函数概述       

            在Flink更底层，我们可以不定义任何具体的算子（比如map，filter，或者window），而只是提炼出一个统一的“处理”（process）操作——它是所有转换算子的一个概括性的表达，可以自定义处理逻辑，所以这一层接口就被叫作“处理函数”（process function）。

     

    二、Process函数分类——8个

    Flink提供了8个不同的处理函数：

    （1）ProcessFunction

    最基本的处理函数，基于DataStream直接调用.process()时作为参数传入。

    （2）KeyedProcessFunction

    对流按键分区后的处理函数，基于KeyedStream调用.process()时作为参数传入。要想使用定时器，比如基于KeyedStream。

    （3）ProcessWindowFunction

    开窗之后的处理函数，也是全窗口函数的代表。基于WindowedStream调用.process()时作为参数传入。

    （4）ProcessAllWindowFunction

    同样是开窗之后的处理函数，基于AllWindowedStream调用.process()时作为参数传入。

    （5）CoProcessFunction

    合并（connect）两条流之后的处理函数，基于ConnectedStreams调用.process()时作为参数传入。关于流的连接合并操作，我们会在后续章节详细介绍。

    （6）ProcessJoinFunction

    间隔连接（interval join）两条流之后的处理函数，基于IntervalJoined调用.process()时作为参数传入。

    （7）BroadcastProcessFunction

    广播连接流处理函数，基于BroadcastConnectedStream调用.process()时作为参数传入。这里的“广播连接流”BroadcastConnectedStream，是一个未keyBy的普通DataStream与一个广播流（BroadcastStream）做连接（conncet）之后的产物。关于广播流的相关操作，我们会在后续章节详细介绍。

    （8）KeyedBroadcastProcessFunction

    按键分区的广播连接流处理函数，同样是基于BroadcastConnectedStream调用.process()时作为参数传入。与BroadcastProcessFunction不同的是，这时的广播连接流，是一个KeyedStream与广播流（BroadcastStream）做连接之后的产物。

    三、KeyedProcessFunction案例

            基于keyBy之后的KeyedStream，直接调用.process()方法，这时需要传入的参数就是KeyedProcessFunction的实现类。

    1. stream.keyBy( t -> t.f0 )
    2.        .process(new MyKeyedProcessFunction())

            类似地，KeyedProcessFunction也是继承自AbstractRichFunction的一个抽象类，与ProcessFunction的定义几乎完全一样，区别只是在于类型参数多了一个K，这是当前按键分区的key的类型。同样地，我们必须实现一个.processElement()抽象方法，用来处理流中的每一个数据；另外还有一个非抽象方法.onTimer()，用来定义定时器触发时的回调操作。

    代码示例：

    1. import com.atguigu.bean.WaterSensor;
    2. import com.atguigu.split.WaterSensorMapFunction;
    3. import org.apache.commons.lang3.time.DateFormatUtils;
    4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    5. import org.apache.flink.streaming.api.TimerService;
    6. import org.apache.flink.streaming.api.datastream.KeyedStream;
    7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    9. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    10. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
    11. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
    12. import org.apache.flink.streaming.api.windowing.time.Time;
    13. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    14. import org.apache.flink.util.Collector;
    15.  
    16. import java.time.Duration;
    17.  
    18. public class KeyedProcessTimerDemo {
    19.     public static void main(String[] args) throws Exception {
    20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    21.         env.setParallelism(1);
    22.  
    23.         SingleOutputStreamOperator sensorDS = env
    24.                 .socketTextStream("node141", 9999)
    25.                 .map(new WaterSensorMapFunction());
    26.  
    27.         WatermarkStrategy watermarkStrategy = WatermarkStrategy
    28.                 .forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 乱序
    29.                 // 提取watermark的时间戳
    30.                 .withTimestampAssigner((element, recordTimestamp) ->
    31.                         element.getTs() * 1000L
    32.                 );
    33.  
    34.         SingleOutputStreamOperator sensorDSWithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);
    35.  
    36.         KeyedStream sensorKS = sensorDSWithWatermark.keyBy(WaterSensor::getId);
    37.         // key    输入的类型     输出的类型
    38.         SingleOutputStreamOperator process = sensorKS.process(
    39.                 new KeyedProcessFunction() {
    40.                     /**
    41.                      * 来一条数据调用一次
    42.                      * @param value
    43.                      * @param ctx
    44.                      * @param out
    45.                      * @throws Exception
    46.                      */
    47.                     @Override
    48.                     public void processElement(WaterSensor value, Context ctx, Collector out) throws Exception {
    49.                         //获取当前数据的key
    50.                         String currentKey = ctx.getCurrentKey();
    51.  
    52.                         // TODO 1.定时器注册
    53.                         TimerService timerService = ctx.timerService();
    54.  
    55.                         // 1、事件时间的案例
    56. //                        Long currentEventTime = ctx.timestamp(); // 数据中提取出来的事件时间
    57. //                        timerService.registerEventTimeTimer(5000L);
    58. //                        System.out.println("当前key=" + currentKey + ",当前时间=" + currentEventTime + ",注册了一个5s的定时器");
    59.  
    60.                         // 2、处理时间的案例
    61.                         long currentTs = timerService.currentProcessingTime();
    62.                         timerService.registerProcessingTimeTimer(currentTs + 5000L);
    63.                         System.out.println("当前key=" + currentKey + ",当前时间=" + currentTs + ",注册了一个5s后的定时器");
    64.  
    65.                         // 3、获取 process的 当前watermark
    66. //                        long currentWatermark = timerService.currentWatermark();
    67. //                        System.out.println("当前数据=" + value + ",当前watermark=" + currentWatermark);
    68.  
    69.  
    70.                         // 注册定时器： 处理时间、事件时间
    71. //                        timerService.registerProcessingTimeTimer();
    72. //                        timerService.registerEventTimeTimer();
    73.  
    74.                         // 删除定时器： 处理时间、事件时间
    75. //                        timerService.deleteEventTimeTimer();
    76. //                        timerService.deleteProcessingTimeTimer();
    77.  
    78.                         // 获取当前时间进展： 处理时间-当前系统时间，  事件时间-当前watermark
    79. //                        long currentTs = timerService.currentProcessingTime();
    80. //                        long wm = timerService.currentWatermark();
    81.                     }
    82.  
    83.                     /**
    84.                      * TODO 2.时间进展到定时器注册的时间，调用该方法
    85.                      * @param timestamp 当前时间进展，就是定时器被触发时的时间
    86.                      * @param ctx       上下文
    87.                      * @param out       采集器
    88.                      * @throws Exception
    89.                      */
    90.                     @Override
    91.                     public void onTimer(long timestamp, OnTimerContext ctx, Collector out) throws Exception {
    92.                         super.onTimer(timestamp, ctx, out);
    93.                         String currentKey = ctx.getCurrentKey();
    94.  
    95.                         System.out.println("key=" + currentKey + "现在时间是" + timestamp + "定时器触发");
    96.                     }
    97.                 }
    98.         );
    99.  
    100.         process.print();
    101.  
    102.         env.execute();
    103.     }
    104. }

    1.运行processElement方法中的事件时间

    1. // 1、事件时间的案例
    2. Long currentEventTime = ctx.timestamp(); // 数据中提取出来的事件时间
    3. timerService.registerEventTimeTimer(5000L);
    4. System.out.println("当前key=" + currentKey + ",当前时间=" + currentEventTime + ",注册了一个5s的定时器");
    (1)输入数据
    1. [root@node141 ~]# nc -lk 9999
    2. s1,1,1
    3. s1,2,2
    4. s1,3,3
    5. s2,4,4
    6. s2,5,5
    7. s3,9,9

    运行结果：注册定时器后，5s后触发

     

    2.运行processElement方法中的处理时间

    1. // 2、处理时间的案例
    2. long currentTs = timerService.currentProcessingTime();
    3. timerService.registerProcessingTimeTimer(currentTs + 5000L);
    4. System.out.println("当前key=" + currentKey + ",当前时间=" + currentTs + ",注册了一个5s后的定时器");
    (1)先输入一条数据
    1. [root@node141 ~]# nc -lk 9999
    2. s1,1,1

    运行结果：相差5s

    (2)再快速输入两条数据
    1. [root@node141 ~]# nc -lk 9999
    2. s1,1,1
    3. s1,2,2
    4. s2,2,2

    运行结果：相差5s，5s后定时器触发

    3.运行processElement方法中的水位线

    1. // 3、获取 process的 当前watermark
    2. long currentWatermark = timerService.currentWatermark();
    3. System.out.println("当前数据=" + value + ",当前watermark=" + currentWatermark);
    (1)输入数据
    1. [root@node141 ~]# nc -lk 9999
    2. s1,1,1
    3. s1,5,5
    4. s1,9,9

    运行结果：

    数据经过map算子，然后再到process方法中。

            当第一条数据输入时，进入map时的watermark是1s-3s-1ms=-2001ms，数据随后进入process方法，调用了processElement方法，在processElement方法中获取当前的watermark，此时-2001ms这一watermark还没有进入process中，所以当前process的watermark是long的最小值；

            第一条数据处理完成后，第二条继续输入到map中，此时process中的watermark变为-2001ms，而map中的watermark是5s-3s-1ms=1999ms，同样，数据再进入process中，调用processElement方法，此时1999ms这一watermark仍然还没有进入process中，所以当前process中的watermark是之前的-2001ms；

           第二条数据处理完成后，第三条数据继续输入到map中，此时process中的watermark变为1999ms，而map中的watermark是9s-3s-1ms=5999ms，同样，数据再进入process中，调用processElement方法，此时5999ms这一watermark也没有来得及进入process中，所以当前process中的watermark是之前的1999ms。

    4.总结：

    TODO 定时器
    1、keyed才有
    2、事件时间定时器，通过watermark来触发的
          watermark >= 注册的时间
          注意： watermark = 当前最大事件时间 - 等待时间 -1ms， 因为 -1ms，所以会推迟一条数据
           比如， 5s的定时器，
           如果 等待=3s， watermark = 8s - 3s -1ms = 4999ms，不会触发5s的定时器
           需要 watermark = 9s -3s -1ms = 5999ms ，才能去触发 5s的定时器
    3、在process中获取当前watermark，显示的是上一次的watermark
          ==> 因为process还没接收到这条数据对应生成的新watermark

    四、应用案例——求TopN

    (一)思路一：hashmap

    使用所有数据到一起，用hashmap来存储，key=vc,value=count值

    1. import com.atguigu.bean.WaterSensor;
    2. import com.atguigu.split.WaterSensorMapFunction;
    3. import org.apache.commons.lang3.time.DateFormatUtils;
    4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    5. import org.apache.flink.api.java.tuple.Tuple2;
    6. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    8. import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
    9. import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
    10. import org.apache.flink.streaming.api.windowing.time.Time;
    11. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    12. import org.apache.flink.util.Collector;
    13.  
    14. import java.time.Duration;
    15. import java.util.*;
    16.  
    17. public class TopNDemo {
    18.     public static void main(String[] args) throws Exception {
    19.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    20.         env.setParallelism(1);
    21.  
    22.         SingleOutputStreamOperator sensorDS = env
    23.                 .socketTextStream("node141", 9999)
    24.                 .map(new WaterSensorMapFunction())
    25.                 .assignTimestampsAndWatermarks(
    26.                         WatermarkStrategy
    27.                                 .forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 乱序
    28.                                 .withTimestampAssigner((element, recordTimestamp) ->
    29.                                         element.getTs() * 1000L
    30.                                 ));
    31.  
    32.         // 最近10s=窗口长度  每5s输出=滑动步长
    33.         // TODO 思路一：使用所有数据到一起，用hashmap来存储，key=vc,value=count值
    34.  
    35.         sensorDS.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    36.                 .process(new MyTopNPawf())
    37.                 .print();
    38.  
    39.         env.execute();
    40.     }
    41.  
    42.     public static class MyTopNPawf extends ProcessAllWindowFunction {
    43.  
    44.         @Override
    45.         public void process(
    46.                 ProcessAllWindowFunction.Context context, Iterable elements, Collector out) throws Exception {
    47.             // 定义一个hashmap用来存，key=vc,value=count值
    48.             Map vcCountMap = new HashMap<>();
    49.             // 1.遍历数据，统计各个vc出现的次数
    50.             for (WaterSensor element : elements) {
    51.                 Integer vc = element.getVc();
    52.                 if (vcCountMap.containsKey(vc)) {
    53.                     // 1.1 key存在，不是这个key的第一条数据，直接累加
    54.                     vcCountMap.put(vc, vcCountMap.get(vc) + 1);
    55.                 } else {
    56.                     // 1.2 如果key不存在，则初始化
    57.                     vcCountMap.put(vc, 1);
    58.                 }
    59.             }
    60.  
    61.             // 2.对count值进行排序，利用list来实现排序
    62.             List> datas = new ArrayList<>();
    63.             // 2.1 对list进行排序，根据count值进行降序
    64.             for (Integer vc : vcCountMap.keySet()) {
    65.                 datas.add(Tuple2.of(vc, vcCountMap.get(vc)));
    66.             }
    67.             // count值相减
    68.             datas.sort((o1, o2) -> o2.f1 - o1.f1);
    69.  
    70.             // 3.取出count最大的2个vc
    71.             StringBuffer outStr = new StringBuffer();
    72.  
    73.             outStr.append("\n=======================\n");
    74.             for (int i = 0; i < Math.min(2, datas.size()); i++) {
    75.                 Tuple2 vcCount = datas.get(i);
    76.                 outStr.append("窗口开始时间：" + DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss.SSS"));
    77.                 outStr.append("\n");
    78.                 outStr.append("top:" + (i + 1));
    79.                 outStr.append("\n");
    80.                 outStr.append("vc=" + vcCount.f0);
    81.                 outStr.append("\n");
    82.                 outStr.append("count=" + vcCount.f1);
    83.                 outStr.append("\n");
    84.                 outStr.append("窗口结束时间：" + DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss.SSS\n"));
    85.             }
    86.             outStr.append("\n=======================\n");
    87.  
    88.             // 输出
    89.             out.collect(outStr.toString());
    90.         }
    91.     }
    92. }

    输入数据：

    1. [root@node141 ~]# nc -lk 9999
    2. s1,1,1
    3. s1,2,1
    4. s1,3,3
    5. s1,6,1
    6. s1,8,2
    7. s1,9,3
    8. s1,10,2
    9. s1,11,1
    10. s1,13,2

    运行结果：

     

    思路一不推荐，因为要将数据攒到一起才会计算，效果不好。

    (二)思路二：keyby

    代码思路：

     

    实现步骤：
    1、按照vc做keyby，开窗，分别count
       ==》 增量聚合，计算 count
       ==》 全窗口，对计算结果 count值封装 ，  带上 窗口结束时间的 标签
             ==》 为了让同一个窗口时间范围的计算结果到一起去
    2、对同一个窗口范围的count值进行处理： 排序、取前N个
       =》 按照 windowEnd做keyby
       =》 使用process， 来一条调用一次，需要先存，分开存，用HashMap进行存储，key=windowEnd,value=List
         =》 使用定时器，对 存起来的结果 进行 排序、取前N个

    1. import com.atguigu.bean.WaterSensor;
    2. import com.atguigu.split.WaterSensorMapFunction;
    3. import org.apache.commons.lang3.time.DateFormatUtils;
    4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    5. import org.apache.flink.api.common.functions.AggregateFunction;
    6. import org.apache.flink.api.java.tuple.Tuple3;
    7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    9. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    10. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
    11. import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
    12. import org.apache.flink.streaming.api.windowing.time.Time;
    13. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    14. import org.apache.flink.util.Collector;
    15.  
    16. import java.time.Duration;
    17. import java.util.ArrayList;
    18. import java.util.HashMap;
    19. import java.util.List;
    20. import java.util.Map;
    21.  
    22. public class KeyedProcessFunctionTopNDemo {
    23.     public static void main(String[] args) throws Exception {
    24.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    25.         env.setParallelism(1);
    26.  
    27.         SingleOutputStreamOperator sensorDS = env
    28.                 .socketTextStream("node141", 9999)
    29.                 .map(new WaterSensorMapFunction())
    30.                 .assignTimestampsAndWatermarks(
    31.                         WatermarkStrategy
    32.                                 .forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 乱序
    33.                                 .withTimestampAssigner((element, recordTimestamp) ->
    34.                                         element.getTs() * 1000L
    35.                                 ));
    36.  
    37.         // 最近10s=窗口长度  每5s输出=滑动步长
    38.         // TODO 使用keyedProcessFunction实现
    39.  
    40.         // 1.按照vc分组、开窗、聚合(增量计算+全量打标签)
    41.         // 开窗聚合后，就是普通的流，没有了窗口信息，需要自己打上窗口的标记 windowEnd
    42.         SingleOutputStreamOperator> windowAgg = sensorDS
    43.                 .keyBy(WaterSensor::getVc)
    44.                 .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    45.                 .aggregate(
    46.                         new VcCountAgg(),
    47.                         new WindowResult()
    48.                 );
    49.  
    50.         // 2.按照窗口标签(窗口结束时间)keyby,保证同一个窗口时间范围的结果，到一起去，排序、取TopN
    51.         windowAgg
    52.                 .keyBy(r -> r.f2)
    53.                 .process(new TopN(2))
    54.                 .print();
    55.  
    56.         env.execute();
    57.     }
    58.  
    59.     public static class VcCountAgg implements AggregateFunction {
    60.  
    61.         @Override
    62.         public Integer createAccumulator() {
    63.             return 0;
    64.         }
    65.  
    66.         @Override
    67.         public Integer add(WaterSensor value, Integer accumulator) {
    68.             return accumulator + 1;
    69.         }
    70.  
    71.         @Override
    72.         public Integer getResult(Integer accumulator) {
    73.             return accumulator;
    74.         }
    75.  
    76.         @Override
    77.         public Integer merge(Integer a, Integer b) {
    78.             return null;
    79.         }
    80.     }
    81.  
    82.     /**
    83.      * 泛型如下：
    84.      * 第一个：输入类型=增量函数的输出 count值，Integer
    85.      * 第二个：输出类型=Tuple3，带上 窗口结束时间的标签
    86.      * 第三个：key的类型：vc,Integer
    87.      * 第四个：窗口类型
    88.      */
    89.     public static class WindowResult extends ProcessWindowFunction, Integer, TimeWindow> {
    90.         @Override
    91.         public void process(Integer key, Context context, Iterable elements, Collector> out) throws Exception {
    92.             // 迭代器里面只有一条数据，next一次即可
    93.             Integer count = elements.iterator().next();
    94.             long windowEnd = context.window().getEnd();
    95.             out.collect(Tuple3.of(key, count, windowEnd));
    96.         }
    97.     }
    98.  
    99.     /**
    100.      * 泛型如下：
    101.      * 第一个：key的类型，是windowEnd
    102.      * 第二个：输入的类型，三元组Tuple3，带上 窗口结束时间的标签
    103.      * 第三个：打印输出结果
    104.      */
    105.     public static class TopN extends KeyedProcessFunction, String> {
    106.         private Map>> dataListMap;
    107.  
    108.         // 要取的top数量
    109.         private int threshold;
    110.  
    111.         public TopN(int threshold) {
    112.             this.threshold = threshold;
    113.             dataListMap = new HashMap<>();
    114.         }
    115.  
    116.         @Override
    117.         public void processElement(Tuple3 value, KeyedProcessFunction, String>.Context ctx, Collector out) throws Exception {
    118.             // 进入这个方法，只是一条数据，要排序，必须等数据到齐才行，将不同窗口的数据用hashmap分开存起来，
    119.             // todo  1.存到hashmap中  注意此时的key是窗口的结束时间
    120.             Long windowEnd = value.f2;
    121.             if (dataListMap.containsKey(windowEnd)) {
    122.                 // 1.1 包含vc,不是该vc的第一条，直接添加到list中
    123.                 List> dataList = dataListMap.get(windowEnd);
    124.                 dataList.add(value);// 也就是说只需要把value的值添加到dataListMap中即可
    125.             } else {
    126.                 // 1.2 不包含，是该vc的第一条，需要初始化list
    127.                 List> dataList = new ArrayList<>();
    128.                 dataList.add(value);
    129.                 dataListMap.put(windowEnd, dataList);
    130.             }
    131.  
    132.             // todo 2.注册一个定时器，windowEnd+1ms即可
    133.             // 同一个窗口范围，应该同时输出
    134.             ctx.timerService().registerEventTimeTimer(windowEnd + 1);
    135.         }
    136.  
    137.         @Override
    138.         public void onTimer(long timestamp, KeyedProcessFunction, String>.OnTimerContext ctx, Collector out) throws Exception {
    139.             super.onTimer(timestamp, ctx, out);
    140.             // 定时器触发，同一个窗口范围的计算结果攒齐了，开始  排序、取TopN
    141.             Long windowEnd = ctx.getCurrentKey();
    142.             // 1.排序
    143.             List> dataList = dataListMap.get(windowEnd);
    144.             dataList.sort((o1, o2) -> o2.f1 - o1.f1);
    145.  
    146.             // 2.取TopN
    147.             StringBuffer outStr = new StringBuffer();
    148.  
    149.             outStr.append("\n=======================\n");
    150.             for (int i = 0; i < Math.min(threshold, dataList.size()); i++) {
    151.                 Tuple3 vcCount = dataList.get(i);
    152.                 outStr.append("top:" + (i + 1));
    153.                 outStr.append("\n");
    154.                 outStr.append("vc=" + vcCount.f0);
    155.                 outStr.append("\n");
    156.                 outStr.append("count=" + vcCount.f1);
    157.                 outStr.append("\n");
    158.                 outStr.append("窗口结束时间：" + DateFormatUtils.format(vcCount.f2, "yyyy-MM-dd HH:mm:ss.SSS" + "\n"));
    159.                 outStr.append("=======================\n");
    160.             }
    161.  
    162.             // 用完的List,及时清理
    163.             dataList.clear();
    164.  
    165.             // 输出
    166.             out.collect(outStr.toString());
    167.         }
    168.     }
    169. }

    输入数据：

    1. [root@node141 ~]# nc -lk 9999
    2. s1,1,1
    3. s1,2,2
    4. s1,8,1
    5. s1,10,1

    运行结果：

    (三)思路三：使用侧输出流——推荐

    1. import com.atguigu.bean.WaterSensor;
    2. import com.atguigu.split.WaterSensorMapFunction;
    3. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    4. import org.apache.flink.api.common.typeinfo.Types;
    5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    7. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    8. import org.apache.flink.util.Collector;
    9. import org.apache.flink.util.OutputTag;
    10.  
    11. import java.time.Duration;
    12.  
    13. public class SideOutputDemo {
    14.     public static void main(String[] args) throws Exception {
    15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    16.         env.setParallelism(1);
    17.  
    18.         SingleOutputStreamOperator sensorDS = env
    19.                 .socketTextStream("node141", 9999)
    20.                 .map(new WaterSensorMapFunction())
    21.                 .assignTimestampsAndWatermarks(
    22.                         WatermarkStrategy
    23.                                 .forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 乱序
    24.                                 .withTimestampAssigner((element, recordTimestamp) ->
    25.                                         element.getTs() * 1000L
    26.                                 ));
    27.  
    28.         OutputTag warnTag = new OutputTag<>("warn", Types.STRING);
    29.         SingleOutputStreamOperator process = sensorDS.keyBy(WaterSensor::getId)
    30.                 .process(new KeyedProcessFunction() {
    31.                     @Override
    32.                     public void processElement(WaterSensor value, KeyedProcessFunction.Context ctx, Collector out) throws Exception {
    33.                         // 使用侧输出流告警
    34.                         if (value.getVc() > 10) {
    35.                             ctx.output(warnTag, "当前水位=" + value.getVc() + ",大于阈值10!!!");
    36.                         }
    37.  
    38.                         // 主流正常发送数据
    39.                         out.collect(value);
    40.                     }
    41.                 });
    42.         process.print("主流");
    43.  
    44.         process.getSideOutput(warnTag).printToErr("warn");
    45.  
    46.         env.execute();
    47.     }
    48. }

    输入数据：

    1. [root@node141 ~]# nc -lk 9999
    2. s1,1,5
    3. s1,10,1
    4. s1,6,11
    5. s1,7,20

    运行结果：

  • 相关阅读:
    你知道怎么改善Java代码吗？
    基于深度学习的图像去雨去雾
    大数据开发之小文件合并
    流程控制.
    解决json-editor-vue3光标错位问题
    基础 | 并发编程 - [Callable & FutureTask]
    简单工厂和工厂方法模式
    5个最好的乐高设计软件
    基于FPGA的超声波测距
    汽车标定技术(九)--标定常量与#pragma的趣事
  • 原文地址：https://blog.csdn.net/Helen_1997_1997/article/details/133604049