在 Flink 中重新分配时间戳、水印？

Question

考虑我这样做：

DataStream<POJO> ds = ...
ds.assignTimestampsAndWatermarks(CustomAssigner)
.windowAll(...)
.apply(someFunction) //THIS FUNCTION CHANGES THE TIMESTAMP FIELD IN THE EVENTS
.assignTimestampsAndWatermarks(AnotherCustomAssigner)

这有效吗？我不知道水印/时间戳是全局的还是仅保留在数据流中？

编辑

class POJO{
   int timestamp;
   String someDetail; //key by this
   ...
}

数据流 ds = ....

ds.assignTimeStampsAndWatermarks(new AssignerWithPunctuatedWatermarks(){
   long maxTS = Long.MIN_VALUE;
   Watermarks checkAndGetNextWater(POJO, p, long l){
  maxTS = max(...)
  return new Watermarks(maxTS);
}

long ExtractTS(POJO p, long l){
  maxTS = max(...)
  return p.timeStamp;
}


  }).keyBy(someDetail property)
     .window(TumblingWindow(1 min))
      .apply(new AllWindowFunction<POJO, POJO, String, TimeWindow>(){
  public void apply(...){
    POJO newPOJO = ...;
    for(POJO it : iterable){
      newPOJO.timeStamp += ...
    }
    collector.collect(newPOJO);
  }
}) 

现在我想知道

如果我应该再次分配时间戳，因为我想做windowAll，然后再次apply。

assignTimestamp...
.windowAll(..)
.apply(some other allwindow function)

Answer 1

您不应再次调用assignTimestampsAndWatermarks。 Flink 将忽略由 WindowFunction 创建的 POJO 中的时间戳，并将使用从该窗口结束时的时间派生的时间戳为包装这些事件的流记录添加时间戳。通常这可以正常工作，尽管后续窗口需要覆盖一个时间范围，该时间范围是第一个时间范围的整数倍。

如果您尝试构建一个全新的流，该流应该重新加时间戳并拥有自己的新水印，那么再次调用 assignTimestampsAndWatermarks 可能会奏效。