第一章.WaterMark

1.WaterMark API

一.升序API

为什么数据没有乱序的情况下还要使用WaterMark

因为只要程序的时间语义是事件时间,就必须使用WaterMark

升序条件下,WaterMark = EventTime(截止目前最大事件时间) - 1(毫秒)

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.datastream.WindowedStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  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;
  16. import java.util.Arrays;
  18. public class $01_WaterMarkMoNo {
  19.     public static void main(String[] args) throws Exception {
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         env.setParallelism(1);
  23.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  24.                 .map(new MapFunction<String, WaterSensor>() {
  25.                     @Override
  26.                     public WaterSensor map(String value) throws Exception {
  27.                         String[] line = value.split(",");
  28.                         return new WaterSensor(
  29.                                 line[0],
  30.                                 Long.parseLong(line[1]),
  31.                                 Integer.parseInt(line[2])
  32.                         );
  34.                     }
  35.                 })
  36.                 .assignTimestampsAndWatermarks(
  37.                         WatermarkStrategy
  38.                         .<WaterSensor>forMonotonousTimestamps()
  39.                         .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  40.                             @Override
  41.                             public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  42.                                 return element.getTs() * 1000L;
  43.                             }
  44.                         })
  45.                 );
  48.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  49.                 keyBy(sensor -> sensor.getId())
  50.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)));
  53.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  54.             @Override
  55.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  56.                 out.collect("key" + s + "\n"
  57.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  58.                         + "数据为:" + Arrays.asList(elements).toString() + "\n"
  59.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  60.             }
  61.         });
  63.         resultDS.print();
  64.         env.execute();
  66.     }
  67. }

二.乱序API

窗口触发条件:

WaterMark = EventTime(截止目前最大事件时间) - maxOutOfOrderness(等待时间) - 1(毫秒)

WaterMark >= maxTimestamp(窗口最大时间戳) 

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.datastream.WindowedStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  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;
  16. import java.time.Duration;
  17. import java.util.Arrays;
  19. public class $02_WaterMarkOutOfOrderness {
  20.     public static void main(String[] args) throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         env.setParallelism(1);
  24.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  25.                 .map(new MapFunction<String, WaterSensor>() {
  26.                     @Override
  27.                     public WaterSensor map(String value) throws Exception {
  28.                         String[] line = value.split(",");
  29.                         return new WaterSensor(
  30.                                 line[0],
  31.                                 Long.parseLong(line[1]),
  32.                                 Integer.parseInt(line[2])
  33.                         );
  35.                     }
  36.                 })
  37.                 .assignTimestampsAndWatermarks(
  38.                         WatermarkStrategy
  39.                          /**
  40.                              *  乱序程度: 1.靠经验值: 2.靠抽样估算
  41.                              *  生产环境: 秒级或者分钟级
  42.                              */
  43.                         .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(4))
  44.                         .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  45.                             @Override
  46.                             public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  47.                                 return element.getTs() * 1000L;
  48.                             }
  49.                         })
  50.                 );
  53.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  54.                 keyBy(sensor -> sensor.getId())
  55.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)));
  58.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  59.             @Override
  60.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  61.                 out.collect("key" + s + "\n"
  62.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  63.                         + "数据为:" + Arrays.asList(elements).toString() + "\n"
  64.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  65.             }
  66.         });
  68.         resultDS.print();
  69.         env.execute();
  71.     }
  72. }

2.自定义WaterMark

有两种风格的WaterMark生产方式:periodic(周期性)和punctuated(间歇性),都需要继承接口:WatermarkGenerztor

一.周期性

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.*;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  7. import org.apache.flink.streaming.api.datastream.WindowedStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  10. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  11. import org.apache.flink.streaming.api.windowing.time.Time;
  12. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  13. import org.apache.flink.util.Collector;
  16. import java.util.Arrays;
  18. /**
  19.  * WaterMark自定义------>周期性生成
  20.  */
  22. public class $03_Mygenerator {
  23.     public static void main(String[] args) throws Exception {
  25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         //TODO 指定watermark生成周期,默认是200ms
  28.         env.getConfig().setAutoWatermarkInterval(2000L);
  29.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  30.                 .map(new MapFunction<String, WaterSensor>() {
  31.                     @Override
  32.                     public WaterSensor map(String value) throws Exception {
  33.                         String[] line = value.split(",");
  34.                         return new WaterSensor(
  35.                                 line[0],
  36.                                 Long.parseLong(line[1]),
  37.                                 Integer.parseInt(line[2])
  38.                         );
  40.                     }
  41.                 })
  42.                 .assignTimestampsAndWatermarks(
  43.                         WatermarkStrategy
  44.                                 /**
  45.                                  *  乱序程度: 1.靠经验值: 2.靠抽样估算
  46.                                  *  生产环境: 秒级或者分钟级
  47.                                  */
  48.                         .forGenerator(new WatermarkGeneratorSupplier<WaterSensor>() {
  49.                             @Override
  50.                             public WatermarkGenerator<WaterSensor> createWatermarkGenerator(Context context) {
  51.                                 return new MyPeriodicGenerator<WaterSensor>(3000L);
  52.                             }
  53.                             })
  54.                             .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  55.                                 @Override
  56.                                 public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  57.                                     return element.getTs() * 1000L;
  58.                                 }
  59.                             })
  61.                 );
  64.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  65.                 keyBy(sensor -> sensor.getId())
  66.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)));
  69.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  70.             @Override
  71.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  72.                 out.collect("key" + s + "\n"
  73.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  74.                         + "数据为:" + Arrays.asList(elements).toString() + "\n"
  75.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  76.             }
  77.         });
  79.         resultDS.print();
  80.         env.execute();
  82.     }
  83.     public static class MyPeriodicGenerator<T> implements WatermarkGenerator<T>{
  84.         long maxOutOfOrderness;
  85.         long maxTs;
  87.         public MyPeriodicGenerator(long maxOutOfOrderness) {
  88.             this.maxOutOfOrderness = maxOutOfOrderness;
  89.             this.maxTs = Long.MIN_VALUE + maxOutOfOrderness;
  90.         }
  92.         /**
  93.          * 每来一次数据,调用一次该方法
  94.          * @param event
  95.          * @param eventTimestamp
  96.          * @param output
  97.          */
  98.         @Override
  99.         public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
  100.             System.out.println("onEvent....");
  101.             maxTs = Math.max(maxTs,eventTimestamp);
  103.         }
  105.         /**
  106.          * 每来一个固定周期,调用一次该方法
  107.          * @param output
  108.          */
  109.         @Override
  110.         public void onPeriodicEmit(WatermarkOutput output) {
  111.             System.out.println("onPeriodicEmit");
  112.             output.emitWatermark(new Watermark(maxTs - maxOutOfOrderness));
  114.         }
  115.     }
  116. }

二.间歇性

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.*;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  7. import org.apache.flink.streaming.api.datastream.WindowedStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  10. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  11. import org.apache.flink.streaming.api.windowing.time.Time;
  12. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  13. import org.apache.flink.util.Collector;
  15. import java.util.Arrays;
  17. /**
  18.  * WaterMark自定义------>间歇性生成
  19.  */
  21. public class $04_MygeneratorByPuntucate {
  22.     public static void main(String[] args) throws Exception {
  24.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  25.         env.setParallelism(1);
  26.         //TODO 指定watermark生成周期,默认是200ms
  27.         env.getConfig().setAutoWatermarkInterval(2000L);
  28.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  29.                 .map(new MapFunction<String, WaterSensor>() {
  30.                     @Override
  31.                     public WaterSensor map(String value) throws Exception {
  32.                         String[] line = value.split(",");
  33.                         return new WaterSensor(
  34.                                 line[0],
  35.                                 Long.parseLong(line[1]),
  36.                                 Integer.parseInt(line[2])
  37.                         );
  39.                     }
  40.                 })
  41.                 .assignTimestampsAndWatermarks(
  42.                         WatermarkStrategy
  43.                                 /**
  44.                                  *  乱序程度: 1.靠经验值: 2.靠抽样估算
  45.                                  *  生产环境: 秒级或者分钟级
  46.                                  */
  47.                         .forGenerator(new WatermarkGeneratorSupplier<WaterSensor>() {
  48.                             @Override
  49.                             public WatermarkGenerator<WaterSensor> createWatermarkGenerator(Context context) {
  50.                                 return new MyPunctuateGenerator<WaterSensor>(3000L);
  51.                             }
  52.                             })
  53.                             .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  54.                                 @Override
  55.                                 public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  56.                                     return element.getTs() * 1000L;
  57.                                 }
  58.                             })
  60.                 );
  63.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  64.                 keyBy(sensor -> sensor.getId())
  65.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)));
  68.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  69.             @Override
  70.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  71.                 out.collect("key" + s + "\n"
  72.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  73.                         + "数据为:" + Arrays.asList(elements).toString() + "\n"
  74.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  75.             }
  76.         });
  78.         resultDS.print();
  79.         env.execute();
  81.     }
  82.     public static class MyPunctuateGenerator<T> implements WatermarkGenerator<T>{
  83.         long maxOutOfOrderness;
  84.         long maxTs;
  86.         public MyPunctuateGenerator(long maxOutOfOrderness) {
  87.             this.maxOutOfOrderness = maxOutOfOrderness;
  88.             this.maxTs = Long.MIN_VALUE + maxOutOfOrderness;
  89.         }
  91.         /**
  92.          * 每来一次数据,调用一次该方法
  93.          * @param event
  94.          * @param eventTimestamp
  95.          * @param output
  96.          */
  97.         @Override
  98.         public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
  99.             System.out.println("onEvent....");
  100.             maxTs = Math.max(maxTs,eventTimestamp);
  101.             Watermark watermark = new Watermark(maxTs - maxOutOfOrderness);
  102.             System.out.println("onEvent>>>watermark=" + watermark.getTimestamp());
  103.             output.emitWatermark(watermark);
  105.         }
  107.         /**
  108.          * 每来一个固定周期,调用一次该方法
  109.          * @param output
  110.          */
  111.         @Override
  112.         public void onPeriodicEmit(WatermarkOutput output) {
  113.             System.out.println("onPeriodicEmit");
  115.         }
  116.     }
  117. }

3.WaterMark老版本写法

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.*;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  7. import org.apache.flink.streaming.api.datastream.WindowedStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
  10. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
  11. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  12. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  13. import org.apache.flink.streaming.api.windowing.time.Time;
  14. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  15. import org.apache.flink.util.Collector;
  17. import java.util.Arrays;
  19. /**
  20.  * WaterMark老版本写法
  21.  */
  23. public class $05_WaterMarkOldVersion {
  24.     public static void main(String[] args) throws Exception {
  26.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  27.         env.setParallelism(1);
  28.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  29.                 .map(new MapFunction<String, WaterSensor>() {
  30.                     @Override
  31.                     public WaterSensor map(String value) throws Exception {
  32.                         String[] line = value.split(",");
  33.                         return new WaterSensor(
  34.                                 line[0],
  35.                                 Long.parseLong(line[1]),
  36.                                 Integer.parseInt(line[2])
  37.                         );
  39.                     }
  40.                 })
  41.                 /**
  42.                  *  1.10及以前版本的写法
  43.                  *
  44.                  */
  45.                 .assignTimestampsAndWatermarks(
  46.                         /**
  47.                          *  TODO 升序写法
  48.                          */
  49.                         /*new AscendingTimestampExtractor<WaterSensor>() {
  50.                             @Override
  51.                             public long extractAscendingTimestamp(WaterSensor element) {
  52.                                 return element.getTs() * 1000L;
  53.                             }
  54.                         }*/
  55.                         /**
  56.                          * TODO 降序写法
  57.                          */
  58.                         new BoundedOutOfOrdernessTimestampExtractor<WaterSensor>(Time.seconds(3)) {
  59.                             @Override
  60.                             public long extractTimestamp(WaterSensor element) {
  61.                                 return element.getTs() * 1000L;
  62.                             }
  63.                         }
  65.                 );
  68.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  69.                 keyBy(sensor -> sensor.getId())
  70.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)));
  73.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  74.             @Override
  75.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  76.                 out.collect("key" + s + "\n"
  77.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  78.                         + "数据为:" + Arrays.asList(elements).toString() + "\n"
  79.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  80.             }
  81.         });
  83.         resultDS.print();
  84.         env.execute();
  86.     }
  89. }

4.多并行度下WaterMark的传递

多并行度的条件下,向下游传递WaterMark的时候,总是以最小的那个WaterMark为准,木桶原理

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.datastream.WindowedStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  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;
  16. import java.time.Duration;
  17. import java.util.Arrays;
  19. /**
  20.  * 多并行度下WaterMark的分析
  21.  */
  23. public class $06_WaterMarkMul {
  24.     public static void main(String[] args) throws Exception {
  26.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  27.         env.setParallelism(2);
  28.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  29.                 .map(new MapFunction<String, WaterSensor>() {
  30.                     @Override
  31.                     public WaterSensor map(String value) throws Exception {
  32.                         String[] line = value.split(",");
  33.                         return new WaterSensor(
  34.                                 line[0],
  35.                                 Long.parseLong(line[1]),
  36.                                 Integer.parseInt(line[2])
  37.                         );
  39.                     }
  40.                 })
  41.                 .assignTimestampsAndWatermarks(
  42.                         WatermarkStrategy
  43.                         .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(4))
  44.                         .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  45.                             @Override
  46.                             public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  47.                                 return element.getTs() * 1000L;
  48.                             }
  49.                         })
  50.                 );
  53.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  54.                 keyBy(sensor -> sensor.getId())
  55.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)));
  58.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  59.             @Override
  60.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  61.                 out.collect("key" + s + "\n"
  62.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  63.                         + "数据为:" + Arrays.asList(elements).toString() + "\n"
  64.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  65.             }
  66.         });
  68.         resultDS.print();
  69.         env.execute();
  71.     }
  72. }

day05[Flink流处理高阶编程(中)] - 图1

day05[Flink流处理高阶编程(中)] - 图2

5.迟到数据

一.窗口允许迟到

  1. 当时间进展 >= maxTs,正常触发输出
  2. 当maxTs < 时间进展 < maxTs + 允许迟到时间,每来一条数据,都会触发一次计算输出
  3. 当时间进展 >= maxTs + 允许迟到时间,窗口关闭,再有迟到的数据来,不会处理
  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.datastream.WindowedStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  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;
  16. import java.time.Duration;
  17. import java.util.Arrays;
  19. public class $07_WaterMarkAllowedLateness {
  20.     public static void main(String[] args) throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         env.setParallelism(1);
  24.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  25.                 .map(new MapFunction<String, WaterSensor>() {
  26.                     @Override
  27.                     public WaterSensor map(String value) throws Exception {
  28.                         String[] line = value.split(",");
  29.                         return new WaterSensor(
  30.                                 line[0],
  31.                                 Long.parseLong(line[1]),
  32.                                 Integer.parseInt(line[2])
  33.                         );
  35.                     }
  36.                 })
  37.                 .assignTimestampsAndWatermarks(
  38.                         WatermarkStrategy
  39.                         .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(4))
  40.                         .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  41.                             @Override
  42.                             public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  43.                                 return element.getTs() * 1000L;
  44.                             }
  45.                         })
  46.                 );
  49.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  50.                 keyBy(sensor -> sensor.getId())
  51.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)))
  52.                 .allowedLateness(Time.seconds(3));
  55.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  56.             @Override
  57.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  58.                 out.collect("key" + s + "\n"
  59.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  60.                         + "数据为:" + Arrays.asList(elements).toString() + "\n"
  61.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  62.             }
  63.         });
  65.         resultDS.print();
  66.         env.execute();
  68.     }
  69. }

二.侧输出流

接收窗口关闭之后的迟到数据

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.datastream.WindowedStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  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. import org.apache.flink.util.OutputTag;
  17. import java.time.Duration;
  18. import java.util.Arrays;
  20. public class $08_WaterMarkSideOutput {
  21.     public static void main(String[] args) throws Exception {
  23.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  24.         env.setParallelism(1);
  25.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  26.                 .map(new MapFunction<String, WaterSensor>() {
  27.                     @Override
  28.                     public WaterSensor map(String value) throws Exception {
  29.                         String[] line = value.split(",");
  30.                         return new WaterSensor(
  31.                                 line[0],
  32.                                 Long.parseLong(line[1]),
  33.                                 Integer.parseInt(line[2])
  34.                         );
  36.                     }
  37.                 })
  38.                 .assignTimestampsAndWatermarks(
  39.                         WatermarkStrategy
  40.                         .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(4))
  41.                         .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  42.                             @Override
  43.                             public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  44.                                 return element.getTs() * 1000L;
  45.                             }
  46.                         })
  47.                 );
  49.         /**
  50.          *  Tag要用匿名内部类写法
  51.          *    1.指定泛型
  52.          *    2.加大括号
  53.          */
  54.         OutputTag<WaterSensor> lateTag = new OutputTag<WaterSensor>("lateTag") {};
  55.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  56.                 keyBy(sensor -> sensor.getId())
  57.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)))
  58.                 .sideOutputLateData(lateTag);
  61.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  62.             @Override
  63.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  64.                 out.collect("key" + s + "\n"
  65.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  66.                         + "数据为:" + Arrays.asList(elements).toString() + "\n"
  67.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  68.             }
  69.         });
  71.         resultDS.print();
  72.         //获取侧输出流的数据
  73.         resultDS.getSideOutput(lateTag).print("late");
  74.         env.execute();
  76.     }
  77. }

6.WaterMark总结

  • WaterMark理解
    • 用来衡量事件时间的进展
    • 用来出来乱序问题
    • 是一个特殊的时间戳,从生成位置插入到流里面
    • 单调不减的
    • 用来触发窗口等
    • Flink认为,事件时间小于WaterMark的数据应该都已经处理过了,如果后续还有事件时间小于WaterMark的数据来,称为迟到数据
  • WaterMark的生成方式
    • 周期性(onPeriodicEmit里发射WaterMark):默认200ms
    • 间歇性(onEvent里发射WaterMark)
  • WaterMark的写法,生成逻辑
    • 升序写法: forMonotonousTimestamps (WaterMark = maxTs - 1ms)
    • 乱序写法: forBoundedOutOfOrderness(Duration) (WaterMark = maxTs - Duration - 1ms)
  • WaterMark的传递
    • 从指定位置插入的,向下游传递
    • 一对多(发送者):广播
    • 多对一(接收者):取最小(木桶原理)
    • 多对多:上面两者的结合,拆开来看
  • Flink对乱序和迟到数据的处理方式
    • WaterMark处理乱序
    • 窗口允许迟到
    • 侧输出流

第二章.ProcessFunction API

之前的转换算子是无法访问事件的时间戳信息,和水位线信息的,ProcessFunction则可以用来构建事件驱动的应用以及实现自定义的业务逻辑

1.Process(时间戳)

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.api.common.functions.MapFunction;
  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.ProcessFunction;
  9. import org.apache.flink.util.Collector;
  11. /**
  12.  * 上下文.timestamp取的是事件时间,要求我们要指定时间的提取,否则为null
  13.  */
  15. public class $09_ProcessDemo {
  16.     public static void main(String[] args) throws Exception {
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         env.setParallelism(1);
  19.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  20.                 .map(new MapFunction<String, WaterSensor>() {
  21.                     @Override
  22.                     public WaterSensor map(String value) throws Exception {
  23.                         String[] line = value.split(",");
  24.                         return new WaterSensor(
  25.                                 line[0],
  26.                                 Long.parseLong(line[1]),
  27.                                 Integer.parseInt(line[2])
  28.                         );
  29.                     }
  30.                 })
  31.                 .assignTimestampsAndWatermarks(
  32.                         WatermarkStrategy
  33.                                 .<WaterSensor>forMonotonousTimestamps()
  34.                         .withTimestampAssigner((data,ts)-> data.getTs() * 1000L)
  35.                 );
  37.         sensorDS
  38.                 .process(
  39.                         new ProcessFunction<WaterSensor, String>() {
  40.                             @Override
  41.                             public void processElement(WaterSensor value, Context context, Collector<String> collector) throws Exception {
  42.                                 collector.collect("数据为：" + value.toString() + "\n" +
  43.                                         "timestamp=" + context.timestamp() + "\n\n");
  44.                             }
  45.                         }
  46.                 )
  47.                 .print();
  48.         env.execute();
  49.     }
  50. }

2.Process(侧输出流)

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.streaming.api.datastream.DataStream;
  7. import org.apache.flink.streaming.api.datastream.KeyedStream;
  8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  11. import org.apache.flink.streaming.api.functions.ProcessFunction;
  12. import org.apache.flink.util.Collector;
  13. import org.apache.flink.util.OutputTag;
  15. /**
  16.  * 侧输出流可以报警,实现分流
  17.  */
  19. public class $09_ProcessSideOutput {
  20.     public static void main(String[] args) throws Exception {
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         env.setParallelism(1);
  23.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  24.                 .map(new MapFunction<String, WaterSensor>() {
  25.                     @Override
  26.                     public WaterSensor map(String value) throws Exception {
  27.                         String[] line = value.split(",");
  28.                         return new WaterSensor(
  29.                                 line[0],
  30.                                 Long.parseLong(line[1]),
  31.                                 Integer.parseInt(line[2])
  32.                         );
  33.                     }
  34.                 })
  35.                 .assignTimestampsAndWatermarks(
  36.                         WatermarkStrategy
  37.                                 .<WaterSensor>forMonotonousTimestamps()
  38.                         .withTimestampAssigner((data,ts)-> data.getTs() * 1000L)
  39.                 );
  40.         KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());
  41.         SingleOutputStreamOperator<String> resultDS = sensorKS.process(new KeyedProcessFunction<String, WaterSensor, String>() {
  42.             @Override
  43.             public void processElement(WaterSensor waterSensor, Context context, Collector<String> collector) throws Exception {
  44.                 OutputTag<Integer> outputTag = new OutputTag<Integer>("baojing") {
  45.                 };
  46.                 if (waterSensor.getVc() > 10) {
  47.                     context.output(outputTag, waterSensor.getVc());
  48.                 } else {
  49.                     collector.collect(waterSensor.toString());
  50.                 }
  51.             }
  52.         });
  53.         resultDS.print();
  54.         DataStream<Integer> sideOutput = resultDS.getSideOutput(new OutputTag<Integer>("baojing") {
  55.         });
  56.         sideOutput.print("side");
  58.         env.execute();
  59.     }
  60. }

第三章.定时器

基于处理时间或事件时间处理过一个元素之后,注册一个定时器,然后在指定的时间执行

1.基于处理时间的定时器

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.streaming.api.TimerService;
  7. import org.apache.flink.streaming.api.datastream.DataStream;
  8. import org.apache.flink.streaming.api.datastream.KeyedStream;
  9. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  12. import org.apache.flink.util.Collector;
  13. import org.apache.flink.util.OutputTag;
  15. /**
  16.  * 基于处理时间的定时器
  17.  */
  19. public class $11_TimeServiceDemo1 {
  20.     public static void main(String[] args) throws Exception {
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         env.setParallelism(1);
  23.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  24.                 .map(new MapFunction<String, WaterSensor>() {
  25.                     @Override
  26.                     public WaterSensor map(String value) throws Exception {
  27.                         String[] line = value.split(",");
  28.                         return new WaterSensor(
  29.                                 line[0],
  30.                                 Long.parseLong(line[1]),
  31.                                 Integer.parseInt(line[2])
  32.                         );
  33.                     }
  34.                 })
  35.                 .assignTimestampsAndWatermarks(
  36.                         WatermarkStrategy
  37.                                 .<WaterSensor>forMonotonousTimestamps()
  38.                         .withTimestampAssigner((data,ts)-> data.getTs() * 1000L)
  39.                 );
  40.         KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());
  41.         SingleOutputStreamOperator<String> resultDS = sensorKS.process(new KeyedProcessFunction<String, WaterSensor, String>() {
  42.             @Override
  43.             public void processElement(WaterSensor waterSensor, Context context, Collector<String> collector) throws Exception {
  44.                 TimerService timerService = context.timerService();
  45.                 System.out.println("currentProcessingTime=" + timerService.currentProcessingTime() + "\n"
  46.                 + "currentWatermark=" + timerService.currentWatermark());
  47.                 timerService.registerProcessingTimeTimer(System.currentTimeMillis() + 1000L);
  48.             }
  50.             /**
  51.              * 定时器触发,当时间进展到了注册的时间,调用该方法
  52.              * @param timestamp
  53.              * @param ctx
  54.              * @param out
  55.              * @throws Exception
  56.              */
  57.             @Override
  58.             public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
  59.                 System.out.println("onTimer的timestamp=" + timestamp);
  60.             }
  61.         });
  62.         resultDS.print();
  65.         env.execute();
  66.     }
  67. }

2.基于事件时间的定时器

  1. package com.atguigu.flink.day05;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.streaming.api.TimerService;
  7. import org.apache.flink.streaming.api.datastream.KeyedStream;
  8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  11. import org.apache.flink.util.Collector;
  13. /**
  14.  * 基于处理时间的定时器
  15.  */
  17. public class $12_TimeServiceDemo2 {
  18.     public static void main(String[] args) throws Exception {
  19.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.         env.setParallelism(1);
  21.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  22.                 .map(new MapFunction<String, WaterSensor>() {
  23.                     @Override
  24.                     public WaterSensor map(String value) throws Exception {
  25.                         String[] line = value.split(",");
  26.                         return new WaterSensor(
  27.                                 line[0],
  28.                                 Long.parseLong(line[1]),
  29.                                 Integer.parseInt(line[2])
  30.                         );
  31.                     }
  32.                 })
  33.                 .assignTimestampsAndWatermarks(
  34.                         WatermarkStrategy
  35.                                 .<WaterSensor>forMonotonousTimestamps()
  36.                         .withTimestampAssigner((data,ts)-> data.getTs() * 1000L)
  37.                 );
  38.         KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());
  39.         SingleOutputStreamOperator<String> resultDS = sensorKS.process(new KeyedProcessFunction<String, WaterSensor, String>() {
  40.             @Override
  41.             public void processElement(WaterSensor waterSensor, Context context, Collector<String> collector) throws Exception {
  42.                 TimerService timerService = context.timerService();
  43.                 System.out.println("currentProcessingTime=" + timerService.currentProcessingTime() + "\n"
  44.                 + "currentWatermark=" + timerService.currentWatermark());
  45.                 timerService.registerEventTimeTimer(context.timestamp() + 5000L);
  46.             }
  48.             /**
  49.              * 定时器触发,当时间进展到了注册的时间,调用该方法
  50.              * @param timestamp
  51.              * @param ctx
  52.              * @param out
  53.              * @throws Exception
  54.              */
  55.             @Override
  56.             public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
  57.                 System.out.println("onTimer的timestamp=" + timestamp +"\n"
  58.                 + "watermark=" + ctx.timerService().currentWatermark());
  59.             }
  60.         });
  61.         resultDS.print();
  64.         env.execute();
  65.     }
  66. }

启动socket测试

  1. [atguigu@hadoop102 ~]$ nc -lk 9999
  2. s_1,1,1
  3. s_1,6,1
  4. s_1,7,1

day05[Flink流处理高阶编程(中)] - 图3