イベント時間ベースのスライディングウィンドウの最新のウィンドウのみを評価する

Question

スライディングウィンドウを使用してEvenTimeでイベントを処理したい。スライド間隔は24時間で、インクリメントは30分です。問題は、以下のコードでは、イベントごとに48回の計算が行われていることです。我々のケースでは、イベントが順番に来ているので、最新のウィンドウだけを評価する必要があります。

おかげで、

デヤン

public static void processEventsa(
     DataStream<Tuple2<String, MyEvent>> events) throws Exception { 

    events.assignTimestampsAndWatermarks(new MyWatermark()). 
      keyBy(0). 
      timeWindow(Time.hours(windowSizeHour), Time.seconds(windowSlideSeconds)). 
      apply(new WindowFunction<Tuple2<String, MyEvent>, Tuple2<String, MyEvent>, Tuple, TimeWindow>() { 
       @Override 
       public void apply(Tuple key, TimeWindow window, Iterable<Tuple2<String, MyEvent>> input, 
              Collector<Tuple2<String, MyEvent>> out) throws Exception { 

        for (Tuple2<String, MyEvent> record : input) { 



        } 
       } 
      }); 
} 

public class MyWatermark implements 
     AssignerWithPunctuatedWatermarks<Tuple2<String, MyEvent>> { 

    @Override 
    public long extractTimestamp(Tuple2<String, MyEvent> event, long previousElementTimestamp) { 
     return event.f1.eventTime; 
    } 

    @Override 
    public Watermark checkAndGetNextWatermark(Tuple2<String, MyEvent> event, long previousElementTimestamp) { 
     return new Watermark(event.f1.eventTime); 
    } 
} 

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 

Answer 1

は問題が透かしにありました。 AssignerWithPeriodicWatermarksを使用する必要があります。

public class MyWatermark implements 
     AssignerWithPeriodicWatermarks<Tuple2<String, MyEvent>> { 

    private final long maxTimeLag = 5000; 

    @Override 
    public long extractTimestamp(Tuple2<String, MyEvent> event, long previousElementTimestamp) { 
     try { 
      return event.f1.eventTime; 
     } 
     catch(NullPointerException ex) {} 

     return System.currentTimeMillis() - maxTimeLag; 
    } 

    @Override 
    public Watermark getCurrentWatermark() { 
     return new Watermark(System.currentTimeMillis() - maxTimeLag); 
    } 
}