devbox
繁依Fanyi0
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

快速灵敏的Flink2_slidingprocessingtimewindows parameters must satis

作者：繁依Fanyi0 | 2024-08-23 03:17:28

赞

踩

slidingprocessingtimewindows parameters must satisfy abs(offset) < slide and

flink基础知识

TumblingEventTimeWindows 滚动开窗

  1. package org.apache.flink.streaming.api.windowing.assigners;
  2.  
  3. import org.apache.flink.annotation.PublicEvolving;
  4. import org.apache.flink.api.common.ExecutionConfig;
  5. import org.apache.flink.api.common.typeutils.TypeSerializer;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.windowing.time.Time;
  8. import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
  9. import org.apache.flink.streaming.api.windowing.triggers.Trigger;
  10. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  11.  
  12. import java.util.Collection;
  13. import java.util.Collections;
  14.  
  15. /**
  16.  * A {@link WindowAssigner} that windows elements into windows based on the timestamp of the
  17.  * elements. Windows cannot overlap.
  18.  *
  19.  * <p>For example, in order to window into windows of 1 minute:
  20.  *
  21.  * <pre>{@code
  22.  * DataStream<Tuple2<String, Integer>> in = ...;
  23.  * KeyedStream<Tuple2<String, Integer>, String> keyed = in.keyBy(...);
  24.  * WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed =
  25.  *   keyed.window(TumblingEventTimeWindows.of(Time.minutes(1)));
  26.  * }</pre>
  27.  */
  28. @PublicEvolving
  29. public class TumblingEventTimeWindows extends WindowAssigner<Object, TimeWindow> {
  30.     private static final long serialVersionUID = 1L;
  31.  
  32.     private final long size;
  33.  
  34.     private final long globalOffset;
  35.  
  36.     private Long staggerOffset = null;
  37.  
  38.     private final WindowStagger windowStagger;
  39.  
  40.     protected TumblingEventTimeWindows(long size, long offset, WindowStagger windowStagger) {
  41.         if (Math.abs(offset) >= size) {
  42.             throw new IllegalArgumentException(
  43.                     "TumblingEventTimeWindows parameters must satisfy abs(offset) < size");
  44.         }
  45.  
  46.         this.size = size;
  47.         this.globalOffset = offset;
  48.         this.windowStagger = windowStagger;
  49.     }
  50.  
  51.     @Override
  52.     public Collection<TimeWindow> assignWindows(
  53.             Object element, long timestamp, WindowAssignerContext context) {
  54.         if (timestamp > Long.MIN_VALUE) {
  55.             if (staggerOffset == null) {
  56.                 staggerOffset =
  57.                         windowStagger.getStaggerOffset(context.getCurrentProcessingTime(), size);
  58.             }
  59.             // Long.MIN_VALUE is currently assigned when no timestamp is present
  60.             long start =
  61.                     TimeWindow.getWindowStartWithOffset(
  62.                             timestamp, (globalOffset + staggerOffset) % size, size);
  63.             return Collections.singletonList(new TimeWindow(start, start + size));
  64.         } else {
  65.             throw new RuntimeException(
  66.                     "Record has Long.MIN_VALUE timestamp (= no timestamp marker). "
  67.                             + "Is the time characteristic set to 'ProcessingTime', or did you forget to call "
  68.                             + "'DataStream.assignTimestampsAndWatermarks(...)'?");
  69.         }
  70.     }
  71.  
  72.     @Override
  73.     public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
  74.         return EventTimeTrigger.create();
  75.     }
  76.  
  77.     @Override
  78.     public String toString() {
  79.         return "TumblingEventTimeWindows(" + size + ")";
  80.     }
  81.  
  82.     /**
  83.      * Creates a new {@code TumblingEventTimeWindows} {@link WindowAssigner} that assigns elements
  84.      * to time windows based on the element timestamp.
  85.      *
  86.      * @param size The size of the generated windows.
  87.      * @return The time policy.
  88.      */
  89.     public static TumblingEventTimeWindows of(Time size) {
  90.         return new TumblingEventTimeWindows(size.toMilliseconds(), 0, WindowStagger.ALIGNED);
  91.     }
  92.  
  93.     /**
  94.      * Creates a new {@code TumblingEventTimeWindows} {@link WindowAssigner} that assigns elements
  95.      * to time windows based on the element timestamp and offset.
  96.      *
  97.      * <p>For example, if you want window a stream by hour,but window begins at the 15th minutes of
  98.      * each hour, you can use {@code of(Time.hours(1),Time.minutes(15))},then you will get time
  99.      * windows start at 0:15:00,1:15:00,2:15:00,etc.
  100.      *
  101.      * <p>Rather than that,if you are living in somewhere which is not using UTC±00:00 time, such as
  102.      * China which is using UTC+08:00,and you want a time window with size of one day, and window
  103.      * begins at every 00:00:00 of local time,you may use {@code of(Time.days(1),Time.hours(-8))}.
  104.      * The parameter of offset is {@code Time.hours(-8))} since UTC+08:00 is 8 hours earlier than
  105.      * UTC time.
  106.      *
  107.      * @param size The size of the generated windows.
  108.      * @param offset The offset which window start would be shifted by.
  109.      */
  110.     public static TumblingEventTimeWindows of(Time size, Time offset) {
  111.         return new TumblingEventTimeWindows(
  112.                 size.toMilliseconds(), offset.toMilliseconds(), WindowStagger.ALIGNED);
  113.     }
  114.  
  115.     /**
  116.      * Creates a new {@code TumblingEventTimeWindows} {@link WindowAssigner} that assigns elements
  117.      * to time windows based on the element timestamp, offset and a staggering offset, depending on
  118.      * the staggering policy.
  119.      *
  120.      * @param size The size of the generated windows.
  121.      * @param offset The globalOffset which window start would be shifted by.
  122.      * @param windowStagger The utility that produces staggering offset in runtime.
  123.      */
  124.     @PublicEvolving
  125.     public static TumblingEventTimeWindows of(Time size, Time offset, WindowStagger windowStagger) {
  126.         return new TumblingEventTimeWindows(
  127.                 size.toMilliseconds(), offset.toMilliseconds(), windowStagger);
  128.     }
  129.  
  130.     @Override
  131.     public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
  132.         return new TimeWindow.Serializer();
  133.     }
  134.  
  135.     @Override
  136.     public boolean isEventTime() {
  137.         return true;
  138.     }
  139. }

SlidingEventTimeWindows 滑动开窗

  1. package org.apache.flink.streaming.api.windowing.assigners;
  2.  
  3. import org.apache.flink.annotation.PublicEvolving;
  4. import org.apache.flink.api.common.ExecutionConfig;
  5. import org.apache.flink.api.common.typeutils.TypeSerializer;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.windowing.time.Time;
  8. import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
  9. import org.apache.flink.streaming.api.windowing.triggers.Trigger;
  10. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  11.  
  12. import java.util.ArrayList;
  13. import java.util.Collection;
  14. import java.util.List;
  15.  
  16. /**
  17.  * A {@link WindowAssigner} that windows elements into sliding windows based on the timestamp of the
  18.  * elements. Windows can possibly overlap.
  19.  *
  20.  * <p>For example, in order to window into windows of 1 minute, every 10 seconds:
  21.  *
  22.  * <pre>{@code
  23.  * DataStream<Tuple2<String, Integer>> in = ...;
  24.  * KeyedStream<Tuple2<String, Integer>, String> keyed = in.keyBy(...);
  25.  * WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed =
  26.  *   keyed.window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10)));
  27.  * }</pre>
  28.  */
  29. @PublicEvolving
  30. public class SlidingEventTimeWindows extends WindowAssigner<Object, TimeWindow> {
  31.     private static final long serialVersionUID = 1L;
  32.  
  33.     private final long size;
  34.  
  35.     private final long slide;
  36.  
  37.     private final long offset;
  38.  
  39.     protected SlidingEventTimeWindows(long size, long slide, long offset) {
  40.         if (Math.abs(offset) >= slide || size <= 0) {
  41.             throw new IllegalArgumentException(
  42.                     "SlidingEventTimeWindows parameters must satisfy "
  43.                             + "abs(offset) < slide and size > 0");
  44.         }
  45.  
  46.         this.size = size;
  47.         this.slide = slide;
  48.         this.offset = offset;
  49.     }
  50.  
  51.     @Override
  52.     public Collection<TimeWindow> assignWindows(
  53.             Object element, long timestamp, WindowAssignerContext context) {
  54.         if (timestamp > Long.MIN_VALUE) {
  55.             List<TimeWindow> windows = new ArrayList<>((int) (size / slide));
  56.             long lastStart = TimeWindow.getWindowStartWithOffset(timestamp, offset, slide);
  57.             for (long start = lastStart; start > timestamp - size; start -= slide) {
  58.                 windows.add(new TimeWindow(start, start + size));
  59.             }
  60.             return windows;
  61.         } else {
  62.             throw new RuntimeException(
  63.                     "Record has Long.MIN_VALUE timestamp (= no timestamp marker). "
  64.                             + "Is the time characteristic set to 'ProcessingTime', or did you forget to call "
  65.                             + "'DataStream.assignTimestampsAndWatermarks(...)'?");
  66.         }
  67.     }
  68.  
  69.     public long getSize() {
  70.         return size;
  71.     }
  72.  
  73.     public long getSlide() {
  74.         return slide;
  75.     }
  76.  
  77.     @Override
  78.     public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
  79.         return EventTimeTrigger.create();
  80.     }
  81.  
  82.     @Override
  83.     public String toString() {
  84.         return "SlidingEventTimeWindows(" + size + ", " + slide + ")";
  85.     }
  86.  
  87.     /**
  88.      * Creates a new {@code SlidingEventTimeWindows} {@link WindowAssigner} that assigns elements to
  89.      * sliding time windows based on the element timestamp.
  90.      *
  91.      * @param size The size of the generated windows.
  92.      * @param slide The slide interval of the generated windows.
  93.      * @return The time policy.
  94.      */
  95.     public static SlidingEventTimeWindows of(Time size, Time slide) {
  96.         return new SlidingEventTimeWindows(size.toMilliseconds(), slide.toMilliseconds(), 0);
  97.     }
  98.  
  99.     /**
  100.      * Creates a new {@code SlidingEventTimeWindows} {@link WindowAssigner} that assigns elements to
  101.      * time windows based on the element timestamp and offset.
  102.      *
  103.      * <p>For example, if you want window a stream by hour,but window begins at the 15th minutes of
  104.      * each hour, you can use {@code of(Time.hours(1),Time.minutes(15))},then you will get time
  105.      * windows start at 0:15:00,1:15:00,2:15:00,etc.
  106.      *
  107.      * <p>Rather than that,if you are living in somewhere which is not using UTC±00:00 time, such as
  108.      * China which is using UTC+08:00,and you want a time window with size of one day, and window
  109.      * begins at every 00:00:00 of local time,you may use {@code of(Time.days(1),Time.hours(-8))}.
  110.      * The parameter of offset is {@code Time.hours(-8))} since UTC+08:00 is 8 hours earlier than
  111.      * UTC time.
  112.      *
  113.      * @param size The size of the generated windows.
  114.      * @param slide The slide interval of the generated windows.
  115.      * @param offset The offset which window start would be shifted by.
  116.      * @return The time policy.
  117.      */
  118.     public static SlidingEventTimeWindows of(Time size, Time slide, Time offset) {
  119.         return new SlidingEventTimeWindows(
  120.                 size.toMilliseconds(), slide.toMilliseconds(), offset.toMilliseconds());
  121.     }
  122.  
  123.     @Override
  124.     public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
  125.         return new TimeWindow.Serializer();
  126.     }
  127.  
  128.     @Override
  129.     public boolean isEventTime() {
  130.         return true;
  131.     }
  132. }

assignTimestampsAndWatermarks 水印

  1.   /**
  2.    * Assigns timestamps to the elements in the data stream and generates watermarks to signal
  3.    * event time progress. The given [[WatermarkStrategy is used to create a [[TimestampAssigner]]
  4.    * and [[org.apache.flink.api.common.eventtime.WatermarkGenerator]].
  5.    *
  6.    * For each event in the data stream, the [[TimestampAssigner#extractTimestamp(Object, long)]]
  7.    * method is called to assign an event timestamp.
  8.    *
  9.    * For each event in the data stream, the
  10.    * [[WatermarkGenerator#onEvent(Object, long, WatermarkOutput)]] will be called.
  11.    *
  12.    * Periodically (defined by the [[ExecutionConfig#getAutoWatermarkInterval()]]), the
  13.    * [[WatermarkGenerator#onPeriodicEmit(WatermarkOutput)]] method will be called.
  14.    *
  15.    * Common watermark generation patterns can be found as static methods in the
  16.    * [[org.apache.flink.api.common.eventtime.WatermarkStrategy]] class.
  17.    */
  18.   def assignTimestampsAndWatermarks(watermarkStrategy: WatermarkStrategy[T]): DataStream[T] = {
  19.     val cleanedStrategy = clean(watermarkStrategy)
  20.  
  21.     asScalaStream(stream.assignTimestampsAndWatermarks(cleanedStrategy))
  22.   }

allowedLateness 分布式架构

分布式架构，有可能出现数据的乱序，窗口要关闭的时候，数据还没有到，那么窗口等一会再关闭，解决数据的迟到问题。允许处理迟到的数据。

  1.   /**
  2.     * Sets the allowed lateness to a user-specified value.
  3.     * If not explicitly set, the allowed lateness is [[0L]].
  4.     * Setting the allowed lateness is only valid for event-time windows.
  5.     * If a value different than 0 is provided with a processing-time
  6.     * [[org.apache.flink.streaming.api.windowing.assigners.WindowAssigner]],
  7.     * then an exception is thrown.
  8.     */
  9.   @PublicEvolving
  10.   def allowedLateness(lateness: Time): WindowedStream[T, K, W] = {
  11.     javaStream.allowedLateness(lateness)
  12.     this
  13.   }

sideOutputLateData  侧输出流

上面allowedLateness()之后，发现还有没到的，放在侧输出流。将迟到的数据放入侧输出流。

  1.   /**
  2.    * Send late arriving data to the side output identified by the given [[OutputTag]]. Data
  3.    * is considered late after the watermark has passed the end of the window plus the allowed
  4.    * lateness set using [[allowedLateness(Time)]].
  5.    *
  6.    * You can get the stream of late data using [[DataStream.getSideOutput()]] on the [[DataStream]]
  7.    * resulting from the windowed operation with the same [[OutputTag]].
  8.    */
  9.   @PublicEvolving
  10.   def sideOutputLateData(outputTag: OutputTag[T]): WindowedStream[T, K, W] = {
  11.     javaStream.sideOutputLateData(outputTag)
  12.     this
  13.   }

flink实操

（1）

  1. package nj.zb.kb23.api.windows
  2.  
  3. import java.time.Duration
  4.  
  5. import nj.zb.kb23.source.SensorReading
  6. import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
  7. import org.apache.flink.streaming.api.scala._
  8. import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
  9. import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
  10. import org.apache.flink.streaming.api.windowing.assigners.{SlidingEventTimeWindows, 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
  14.  
  15. object WindowEventTimeTest {
  16.   def main(args: Array[String]): Unit = {
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  19.     val inputStream: DataStream[String] = env.socketTextStream("192.168.91.11", 7777)
  20.     val dataStream: DataStream[SensorReading] = inputStream.map(data => {
  21.       val arr: Array[String] = data.split(",")
  22.       SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
  23.     })
  24.     //    dataStream.setParallelism()
  25.     //设置以事件时间为时间语意
  26.     val dataStream2: DataStream[SensorReading] = dataStream.assignTimestampsAndWatermarks(
  27.       WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds((3)))
  28.         .withTimestampAssigner(
  29.           new SerializableTimestampAssigner[SensorReading] {
  30.             override def extractTimestamp(element: SensorReading, recordTimestamp: Long): Long = {
  31.               //指定事件时间的字段
  32.               element.timestamp * 1000
  33.             }
  34.           }
  35.         )
  36.     )
  37.     val lataTag = new OutputTag[SensorReading]("lastdata")
  38.     val windowStream: WindowedStream[SensorReading, String, TimeWindow] = dataStream2.keyBy(_.id)
  39.       .window(TumblingEventTimeWindows.of(Time.seconds(15))) //滚动开窗
  40. //      .window(SlidingEventTimeWindows.of(Time.seconds(15),Time.seconds(3)))//滑动开窗
  41.       .allowedLateness(Time.minutes(1)) //最长延迟一分钟
  42.       .sideOutputLateData(lataTag)//非常非常延迟的数据,放入侧输出流
  43.  
  44.     /*    val resultStream: DataStream[SensorReading] = windowStream.reduce(
  45.       (curReduce, newReducw) => {
  46.         SensorReading(curReduce.id, curReduce.timestamp, curReduce.temperature.min(newReducw.temperature))
  47.       }
  48.     )*/
  49.    val resultStream: DataStream[SensorReading] = windowStream.process(new MyEventProcessWindowFunction)
  50.     resultStream.print("result：")
  51.     resultStream.getSideOutput(lataTag).print("late value:")//.addSink()
  52.     env.execute("windowEventTime")
  53.  
  54.   }
  55. }
  56.  
  57. class MyEventProcessWindowFunction extends ProcessWindowFunction[SensorReading,SensorReading,String,TimeWindow]{
  58.   override def process(
  59.                         key: String,
  60.                         context: Context,
  61.                         elements: Iterable[SensorReading],
  62.                         out: Collector[SensorReading]): Unit = {
  63.     val window: TimeWindow = context.window
  64.     println(window.getStart,window.getEnd)
  65.     val iterator: Iterator[SensorReading] = elements.iterator
  66.     var temp = 100.0
  67.     var timestam = 1
  68.     out.collect(SensorReading(key,timestam,temp))
  69. //    out.collect(SensorReading(key,1,0.0))
  70.   }
  71. }

（2）

  1. package nj.zb.kb23.api.windows
  2.  
  3. import java.time.Duration
  4.  
  5. import nj.zb.kb23.source.SensorReading
  6. import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
  7. import org.apache.flink.streaming.api.scala._
  8. import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
  9. import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
  10. import org.apache.flink.streaming.api.windowing.assigners.{SlidingEventTimeWindows, 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
  14.  
  15. object WindowEventTimeTest {
  16.   def main(args: Array[String]): Unit = {
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  19.     val inputStream: DataStream[String] = env.socketTextStream("192.168.91.11", 7777)
  20.     val dataStream: DataStream[SensorReading] = inputStream.map(data => {
  21.       val arr: Array[String] = data.split(",")
  22.       SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
  23.     })
  24.     //    dataStream.setParallelism()
  25.     //设置以时间时间为时间语意
  26.     val dataStream2: DataStream[SensorReading] = dataStream.assignTimestampsAndWatermarks(
  27.       WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds((3)))
  28.         .withTimestampAssigner(
  29.           new SerializableTimestampAssigner[SensorReading] {
  30.             override def extractTimestamp(element: SensorReading, recordTimestamp: Long): Long = {
  31.               //指定事件时间的字段
  32.               element.timestamp * 1000
  33.             }
  34.           }
  35.         )
  36.     )
  37.     val lataTag = new OutputTag[SensorReading]("lastdata")
  38.     val windowStream: WindowedStream[SensorReading, String, TimeWindow] = dataStream2.keyBy(_.id)
  39.       .window(TumblingEventTimeWindows.of(Time.seconds(15))) //滚动开窗
  40. //      .window(SlidingEventTimeWindows.of(Time.seconds(15),Time.seconds(3)))//滑动开窗
  41.       .allowedLateness(Time.minutes(1)) //最长延迟一分钟
  42.       .sideOutputLateData(lataTag)//非常非常延迟的数据,放入侧输出流
  43.  
  44.     val resultStream: DataStream[(String, Double, Long, Long)] = windowStream.process(new MyEventProcessWindowFunction)
  45.     resultStream.print("result：")
  46.     resultStream.getSideOutput(lataTag).print("late value:")//.addSink()
  47.     env.execute("windowEventTime")
  48.  
  49.   }
  50. }
  51. class MyEventProcessWindowFunction extends ProcessWindowFunction[SensorReading,(String,Double,Long,Long),String,TimeWindow]{
  52.   override def process(
  53.                         key: String,
  54.                         context: Context,
  55.                         elements: Iterable[SensorReading],
  56.                         out: Collector[(String, Double, Long, Long)]): Unit = {
  57.     val window: TimeWindow = context.window
  58.     val iterator: Iterator[SensorReading] = elements.iterator
  59.     var temp = 100.0
  60.     var timestam = 1
  61.     while (iterator.hasNext){
  62.       val sensorReading: SensorReading = iterator.next()
  63.       temp = temp.min(sensorReading.temperature)
  64.     }
  65.     out.collect((key,temp,window.getStart,window.getEnd))
  66.   }
  67. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/繁依Fanyi0/article/detail/1019226
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号