阅读 5534

Flink Window分析及Watermark解决乱序数据机制深入剖析-Flink牛刀小试

专注于大数据及容器云核心技术解密,可提供全栈的大数据+云原生平台咨询方案,请持续关注本套博客。如有任何学术交流,可随时联系。更多内容请关注《数据云技术社区》公众号。

本文决心讲清楚这个纠结的水印Watermark问题,Come on !

1 The Time

针对stream数据中的时间,可以分为以下三种:

  • Event Time:事件产生的时间,它通常由事件中的时间戳描述。
  • Ingestion time:事件进入Flink的时间
  • Processing Time:事件被处理时当前系统的时间

  • Flink中,默认Time类似是ProcessingTime ,可以在代码中设置:

1.1 tips(请认真思考下面的话,绝对震聋发溃!)

  • 在水印的处理中,我们一般取事件时间序列的最大值作为水印的生成时间参考。

  • 按照信号发生的顺序,时间是不断增加的,所以在时间序列上若出现事件时间小于时间序列最大值,一般都是延时的事件,时间序列最大值不会改变。

  • 每处理一条事件数据,watermark时间就生成一次,后面窗的触发就是依据水印时间。若设置乱序延时为10s,则生成规则就是:

     final Long maxOutOfOrderness = 10000L;// 最大允许的乱序时间是10s
     new Watermark(currentMaxTimestamp - maxOutOfOrderness)
    复制代码
  • 数据会按照时间进行依次Append,

  • 水印依赖的条件是窗,水印只是决定了窗的触发时间,若设置最大允许的乱序时间是maxOutOfOrderness=10s,则窗的触发时机就是:

      watermark 时间 >= window_end_time
    复制代码
  • 窗触发时,数据除了正常的时间序列,同时也包含延时到达的序列。在窗触发前(也就水印时间),计算除了把之前的正常窗数据给触发了,同时还包含了本来也属于该窗的延时数据。

2 窗与水印的世纪谜题

  • 事件时间的最大值,也就是当前的实际事件时间,因此需要以此为参考点。

  • 实际窗:意思就是数据就在那里Append,窗数据已经准备好,等待触发时机。

  • 水印时间不受影响:就是每次来的数据的事件时间最大值,不受延迟数据时间影响。

  • 下面例子中,等水印时间为10:11:33时,满足时间窗 10:11:30 <-> 10:11:33的触发时机,此时需要处理的数据不仅包含正常数据10:11:32 ,同时还包含乱序数据10:11:31。

  • 再次强调:窗时机到来时,会遍历乱序数据和原窗数据。

  • 实际窗在流动,只是暂不触发。

  • 水印也在标记流动

  • 窗时机触发也在流动。

  • watermark 时间 >= window_end_time时,触发历史窗执行。

  • 继续

3 EventTime和Watermarks 水位线理论碰撞

  • 流处理从事件产生,到流经source,再到operator,中间是有一个过程和时间的。虽然大部分情况下,流到operator的数据都是按照事件产生的时间顺序来的,但是也不排除由于网络延迟等原因,导致乱序的产生,特别是使用kafka的话,多个分区的数据无法保证有序。所以在进行window计算的时候,我们又不能无限期的等下去,必须要有个机制来保证一个特定的时间后,必须触发window去进行计算了。这个特别的机制,就是watermark,watermark是用于处理乱序事件的。

  • 通常,在接收到source的数据后,应该立刻生成watermark;但是,也可以在source后,应用简单的map或者filter操作后,再生成watermark。注意:如果指定多次watermark,后面指定的会覆盖前面的值。 生成方式

  • With Periodic Watermarks

      周期性的触发watermark的生成和发送,默认是100ms,每隔N秒自动向流里
      注入一个WATERMARK 时间间隔由ExecutionConfig.setAutoWatermarkInterval 
      决定. 每次调用getCurrentWatermark 方法, 如果得到的WATERMARK
      不为空并且比之前的大就注入流中 
      可以定义一个最大允许乱序的时间,这种比较常用
      实现AssignerWithPeriodicWatermarks接口
    复制代码
  • With Punctuated Watermarks

      基于某些事件触发watermark的生成和发送基于事件向流里注入一个WATERMARK,
      每一个元素都有机会判断是否生成一个WATERMARK. 如果得到的WATERMARK
      不为空并且比之前的大就注入流中实现AssignerWithPunctuatedWatermarks接口
    复制代码
  • 多并行度流的watermarks

    注意:多并行度的情况下,watermark对齐会取所有channel最小的watermark。

4 With Periodic Watermarks案例实战

4.1 最朴实的水印方案(基于事件序列最大值)

 public class StreamingWindowWatermark {

    public static void main(String[] args) throws Exception {
        //定义socket的端口号
        int port = 9010;
        //获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置使用eventtime,默认是使用processtime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置并行度为1,默认并行度是当前机器的cpu数量
        env.setParallelism(1);
        //连接socket获取输入的数据
        DataStream<String> text = env.socketTextStream("SparkMaster", port, "\n");

        //解析输入的数据
        DataStream<Tuple2<String, Long>> inputMap = text.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String value) throws Exception {
                String[] arr = value.split(",");
                return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
            }
        });

        //抽取timestamp和生成watermark
        DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {

            Long currentMaxTimestamp = 0L;
            final Long maxOutOfOrderness = 10000L;// 最大允许的乱序时间是10s

            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
 
            @Nullable
            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
            }

            //定义如何提取timestamp
            @Override
            public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
                long timestamp = element.f1;
                currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                long id = Thread.currentThread().getId();
                System.out.println("作者:秦凯新 键值 :"+element.f0+",事件事件:[ "+sdf.format(element.f1)+" ],currentMaxTimestamp:[ "+
                        sdf.format(currentMaxTimestamp)+" ],水印时间:[ "+sdf.format(getCurrentWatermark().getTimestamp())+" ]");
                return timestamp;
            }
        });

        DataStream<String> window = waterMarkStream.keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(3)))//按照消息的EventTime分配窗口,和调用TimeWindow效果一样
                .apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
                    /**
                     * 对window内的数据进行排序,保证数据的顺序
                     * @param tuple
                     * @param window
                     * @param input
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
                        String key = tuple.toString();
                        List<Long> arrarList = new ArrayList<Long>();
                        Iterator<Tuple2<String, Long>> it = input.iterator();
                        while (it.hasNext()) {
                            Tuple2<String, Long> next = it.next();
                            arrarList.add(next.f1);
                        }
                        Collections.sort(arrarList);
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                        String result = "\n 作者:秦凯新 键值 : "+ key + "\n              触发窗内数据个数 : " + arrarList.size() + "\n              触发窗起始数据: " + sdf.format(arrarList.get(0)) + "\n              触发窗最后(可能是延时)数据:" + sdf.format(arrarList.get(arrarList.size() - 1))
                                + "\n              实际窗起始和结束时间: " + sdf.format(window.getStart()) + "《----》" + sdf.format(window.getEnd()) + " \n \n ";

                        out.collect(result);
                    }
                });
        //测试-把结果打印到控制台即可
        window.print();

        //注意:因为flink是懒加载的,所以必须调用execute方法,上面的代码才会执行
        env.execute("eventtime-watermark");

    }
}
复制代码
  • 执行测试数据

     0001,1538359882000		2018-10-01 10:11:22
     0002,1538359886000		2018-10-01 10:11:26
     0003,1538359892000		2018-10-01 10:11:32
     0004,1538359893000		2018-10-01 10:11:33
     0005,1538359894000		2018-10-01 10:11:34
     0006,1538359896000		2018-10-01 10:11:36
     0007,1538359897000		2018-10-01 10:11:37
     
     0008,1538359899000		2018-10-01 10:11:39
     0009,1538359891000		2018-10-01 10:11:31
     0010,1538359903000		2018-10-01 10:11:43
     
     0011,1538359892000		2018-10-01 10:11:32
     0012,1538359891000		2018-10-01 10:11:31
     
     0010,1538359906000		2018-10-01 10:11:46
    复制代码

第一个窗触发:2018-10-01 10:11:21.000《----》2018-10-01 10:11:24.000

第二个窗触发:2018-10-01 10:11:24.000《----》2018-10-01 10:11:27.000

第三个窗触发:2018-10-01 10:11:30.000《----》2018-10-01 10:11:33.000

第四个窗触发:10:11:33.000《----》2018-10-01 10:11:36.000

4.2 最霸道的水印设计(allowedLateness与OutputLateData)

  • 在某些情况下, 我们希望对迟到的数据再提供一个宽容的时间。 Flink 提供了 allowedLateness 方法可以实现对迟到的数据设置一个延迟时间, 在指定延迟时间内到达的数据还是可以触发 window 执行的。

  • 第二次(或多次)触发的条件是 watermark < window_end_time + allowedLateness 时间内, 这个窗口有 late 数据到达时。

  • 举例:当 watermark 等于 10:11:34 的时候, 我们输入 eventtime 为 10:11:30、 10:11:31、10:11:32 的数据的时候, 是可以触发的, 因为这些数据的 window_end_time 都是 10:11:33, 也就是10:11:34<10:11:33+2 为 true。

  • 举例:但是当 watermark 等于 10:11:35 的时候,我们再输入 eventtime 为 10:11:30、10:11:31、10:11:32的数据的时候, 这些数据的 window_end_time 都是 10:11:33, 此时, 10:11:35< 10:11:33+2 为false 了。 所以最终这些数据迟到的时间太久了, 就不会再触发 window 执行了,预示着丢弃。

  • 同时注意,对于延迟的数据,我们完全可以把它揪出来作分析。通过设置sideOutputLateData。

      public class StreamingWindowWatermark2 {
      public static void main(String[] args) throws Exception {
          //定义socket的端口号
          int port = 9000;
          //获取运行环境
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
          //设置使用eventtime,默认是使用processtime
          env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
          //设置并行度为1,默认并行度是当前机器的cpu数量
          env.setParallelism(1);
    
          //连接socket获取输入的数据
          DataStream<String> text = env.socketTextStream("hadoop100", port, "\n");
    
          //解析输入的数据
          DataStream<Tuple2<String, Long>> inputMap = text.map(new MapFunction<String, Tuple2<String, Long>>() {
              @Override
              public Tuple2<String, Long> map(String value) throws Exception {
                  String[] arr = value.split(",");
                  return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
              }
          });
    
          //抽取timestamp和生成watermark
          DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {
    
              Long currentMaxTimestamp = 0L;
              final Long maxOutOfOrderness = 10000L;// 最大允许的乱序时间是10s
    
              SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
              /**
               * 定义生成watermark的逻辑
               * 默认100ms被调用一次
               */
              @Nullable
              @Override
              public Watermark getCurrentWatermark() {
                  return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
              }
    
              //定义如何提取timestamp
              @Override
              public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
                  long timestamp = element.f1;
                  currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                  System.out.println("key:"+element.f0+",eventtime:["+element.f1+"|"+sdf.format(element.f1)+"],currentMaxTimestamp:["+currentMaxTimestamp+"|"+
                          sdf.format(currentMaxTimestamp)+"],watermark:["+getCurrentWatermark().getTimestamp()+"|"+sdf.format(getCurrentWatermark().getTimestamp())+"]");
                  return timestamp;
              }
          });
    
          //保存被丢弃的数据
          OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("late-data"){};
          //注意,由于getSideOutput方法是SingleOutputStreamOperator子类中的特有方法,所以这里的类型,不能使用它的父类dataStream。
          SingleOutputStreamOperator<String> window = waterMarkStream.keyBy(0)
                  .window(TumblingEventTimeWindows.of(Time.seconds(3)))//按照消息的EventTime分配窗口,和调用TimeWindow效果一样
                  //.allowedLateness(Time.seconds(2))//允许数据迟到2秒
                  .sideOutputLateData(outputTag)
                  .apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
                      /**
                       * 对window内的数据进行排序,保证数据的顺序
                       * @param tuple
                       * @param window
                       * @param input
                       * @param out
                       * @throws Exception
                       */
                      @Override
                      public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
                          String key = tuple.toString();
                          List<Long> arrarList = new ArrayList<Long>();
                          Iterator<Tuple2<String, Long>> it = input.iterator();
                          while (it.hasNext()) {
                              Tuple2<String, Long> next = it.next();
                              arrarList.add(next.f1);
                          }
                          Collections.sort(arrarList);
                          SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                          String result = key + "," + arrarList.size() + "," + sdf.format(arrarList.get(0)) + "," + sdf.format(arrarList.get(arrarList.size() - 1))
                                  + "," + sdf.format(window.getStart()) + "," + sdf.format(window.getEnd());
                          out.collect(result);
                      }
                  });
          //把迟到的数据暂时打印到控制台,实际中可以保存到其他存储介质中
          DataStream<Tuple2<String, Long>> sideOutput = window.getSideOutput(outputTag);
          sideOutput.print();
          //测试-把结果打印到控制台即可
          window.print();
    
          //注意:因为flink是懒加载的,所以必须调用execute方法,上面的代码才会执行
          env.execute("eventtime-watermark");
          }
      }
    复制代码

4.3 多并行度下的 watermark触发机制

4.3.1 先领会代码(感谢 github xuwei)

    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
    import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    import org.apache.flink.streaming.api.watermark.Watermark;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.util.Collector;
    import org.apache.flink.util.OutputTag;
    
    import javax.annotation.Nullable;
    import java.text.SimpleDateFormat;
    import java.util.ArrayList;
    import java.util.Collections;
    import java.util.Iterator;
    import java.util.List;
    
    public class StreamingWindowWatermark2 {
    
        public static void main(String[] args) throws Exception {
            //定义socket的端口号
            int port = 9010;
            //获取运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //设置使用eventtime,默认是使用processtime
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
            //设置并行度为1,默认并行度是当前机器的cpu数量
            env.setParallelism(8);
    
            //连接socket获取输入的数据
            DataStream<String> text = env.socketTextStream("SparkMaster", port, "\n");
    
            //解析输入的数据
            DataStream<Tuple2<String, Long>> inputMap = text.map(new MapFunction<String, Tuple2<String, Long>>() {
                @Override
                public Tuple2<String, Long> map(String value) throws Exception {
                    String[] arr = value.split(",");
                    return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
                }
            });
    
            //抽取timestamp和生成watermark
            DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {
    
                Long currentMaxTimestamp = 0L;
                final Long maxOutOfOrderness = 10000L;// 最大允许的乱序时间是10s
    
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                /**
                 * 定义生成watermark的逻辑
                 * 默认100ms被调用一次
                 */
                @Nullable
                @Override
                public Watermark getCurrentWatermark() {
                    return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
                }
    
                //定义如何提取timestamp
                @Override
                public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
                    long timestamp = element.f1;
                    currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                    long id = Thread.currentThread().getId();
                    System.out.println("作者:秦凯新 键值 :"+element.f0+"线程验证 :"+  id   +" , 事件事件:[ "+sdf.format(element.f1)+" ],currentMaxTimestamp:[ "+
                            sdf.format(currentMaxTimestamp)+" ],水印时间:[ "+sdf.format(getCurrentWatermark().getTimestamp())+" ]");                return timestamp;
                }
            });
    
            //保存被丢弃的数据
            OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("late-data"){};
            //注意,由于getSideOutput方法是SingleOutputStreamOperator子类中的特有方法,所以这里的类型,不能使用它的父类dataStream。
            SingleOutputStreamOperator<String> window = waterMarkStream.keyBy(0)
                    .window(TumblingEventTimeWindows.of(Time.seconds(3)))//按照消息的EventTime分配窗口,和调用TimeWindow效果一样
                    //.allowedLateness(Time.seconds(2))//允许数据迟到2秒
                    .sideOutputLateData(outputTag)
                    .apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
                        /**
                         * 对window内的数据进行排序,保证数据的顺序
                         * @param tuple
                         * @param window
                         * @param input
                         * @param out
                         * @throws Exception
                         */
                        @Override
                        public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
                            String key = tuple.toString();
                            List<Long> arrarList = new ArrayList<Long>();
                            Iterator<Tuple2<String, Long>> it = input.iterator();
                            while (it.hasNext()) {
                                Tuple2<String, Long> next = it.next();
                                arrarList.add(next.f1);
                            }
                            Collections.sort(arrarList);
                            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                            String result = "\n 作者:秦凯新 键值 : "+ key + "\n              触发窗内数据个数 : " + arrarList.size() + "\n              触发窗起始数据: " + sdf.format(arrarList.get(0)) + "\n              触发窗最后(可能是延时)数据:" + sdf.format(arrarList.get(arrarList.size() - 1))
                                    + "\n              实际窗起始和结束时间: " + sdf.format(window.getStart()) + "《----》" + sdf.format(window.getEnd()) + " \n \n ";
                            out.collect(result);
                        }
                    });
            //把迟到的数据暂时打印到控制台,实际中可以保存到其他存储介质中
            DataStream<Tuple2<String, Long>> sideOutput = window.getSideOutput(outputTag);
            sideOutput.print();
            //测试-把结果打印到控制台即可
            window.print();
    
            //注意:因为flink是懒加载的,所以必须调用execute方法,上面的代码才会执行
            env.execute("eventtime-watermark");
    
        }
    }
复制代码

4.3.2 前面代码中设置了并行度为 1:

    env.setParallelism(1);
复制代码

如果这里不设置的话, 代码在运行的时候会默认读取本机 CPU 数量设置并行度。

下面我们来验证一下, 把代码中的并行度调整为 2:

    env.setParallelism(2);
复制代码
  • 发现玄机如下:在第二条事件时,其实已经达到窗的触发时机,但是因为并行度为2,只有等到最小

  • watermark 到的时候才会触发窗计算。发现线程44处理的是001和003 ,线程42处理的是0002,所以只有等到线程42到达后,水印才会起作用执行2018-10-01 10:11:33.000所在的窗。

      0001,1538359890000		2018-10-01 10:11:30
      0002,1538359903000		2018-10-01 10:11:43
      0003,1538359908000		2018-10-01 10:11:48
    复制代码

4.3.3 现在代码中设置了并行度为 8:

  • 发现 这 7 条数据都是被不同的线程处理的。 每个线程都有一个 watermark。且每一个线程都是基于自己接收数据的事件时间最大值。

  • 因此,导致到最后现在还没获取到最小的 watermark, 所以 window 无法被触发执行。

  • 只有所有的线程的最小watermark都满足watermark 时间 >= window_end_time时,触发历史窗才会执行。

      0001,1538359882000		2018-10-01 10:11:22
      0002,1538359886000		2018-10-01 10:11:26
      0003,1538359892000		2018-10-01 10:11:32
      0004,1538359893000		2018-10-01 10:11:33
      0005,1538359894000		2018-10-01 10:11:34
      0006,1538359896000		2018-10-01 10:11:36
      0007,1538359897000		2018-10-01 10:11:37
    复制代码

  • 当持续发生事件数据时。一旦所有线程都达到最低的窗触发时机时,就会进行窗触发执行了。输入数据如下:

      0007,1538359897000		2018-10-01 10:11:37
      0008,1538359897000		2018-10-01 10:11:37
      0009,1538359897000		2018-10-01 10:11:37
      0010,1538359897000		2018-10-01 10:11:37
      0011,1538359897000		2018-10-01 10:11:37
      0012,1538359897000		2018-10-01 10:11:37
      0013,1538359897000		2018-10-01 10:11:37
      0014,1538359897000		2018-10-01 10:11:37
      0015,1538359897000		2018-10-01 10:11:37
    复制代码

5 再下一城

专注于大数据及容器云核心技术解密,可提供全栈的大数据+云原生平台咨询方案,请持续关注本套博客。如有任何学术交流,可随时联系。更多内容请关注《数据云技术社区》公众号。

秦凯新 于深圳