flink事件属于窗口的计算方法

我们都知道flink有窗口和watermark的概念，当watermark大于窗口的endTime，将触发窗口中数据的计算，watermark是一个不断递增的时间戳，是不断变化的，如果我们假设一个窗口的开始时间和结束时间也是不断变化的，那么watermark就不好触发窗口计算。所以根据我们的假设，内心也是认为一个特定的窗口的开始和结束时间肯定是固定的。

如果一个特定的窗口是不会变化的，比如滚动窗口，我们在代码中只需要传入窗口的size，就可以完成窗口的构建，那么窗口的开始时间和结束时间是怎么获取的。

例如我们定义一个窗口：window(TumblingProcessingTimeWindows.of(Time.seconds(10))),当我们当前事件的处理时间是2021-12-20 10:17:14，那么这个事件是属于下面窗口中的哪一个？[10:17:10, 10:17:20)，[10:17:11, 10:17:21)，[10:17:12, 10:17:22)，[10:17:14, 10:17:24)，…（共十种可能）

我们知道如果我们设置了窗口大小，那么这些窗口都是固定不变的，也就是说这些窗口都是真实存在的，不管你用不用它，那么第一个窗口是哪一个呢？
答案就是所有类型的窗口，第一个窗口的开始时间都是1970-01-01 08:00:00

ok，我们来着上面的疑问向下走。。。

下面列出窗口的示例代码

source.map(new MyMapFunction()).setParallelism(8).keyBy(s -> s.f0)
  .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
  .process(new MyProcessWindowFunction()).setParallelism(8).print();

通过上面TumblingProcessingTimeWindows.of(Time.seconds(10))创建是一个窗口大小是10s的窗口。

public static TumblingProcessingTimeWindows of(Time size) {
  return new TumblingProcessingTimeWindows(size.toMilliseconds(), 0, WindowStagger.ALIGNED);
}

也就完成了窗口中三个变量的初始化：

private TumblingProcessingTimeWindows(long size, long offset, WindowStagger windowStagger) {
  if (Math.abs(offset) >= size) {
    throw new IllegalArgumentException(
      "TumblingProcessingTimeWindows parameters must satisfy abs(offset) < size");
  }

  this.size = size;
  this.globalOffset = offset;
  this.windowStagger = windowStagger;
}

可以看到窗口中的size=10000(10秒), globalOffset=0, windowStagger=WindowStagger.ALIGNED(所有窗口都是从0开始)

现在看下TumblingProcessingTimeWindows分配窗口的功能：assignWindows

@Override
public Collection assignWindows(
  Object element, long timestamp, WindowAssignerContext context) {
  final long now = context.getCurrentProcessingTime();
  if (staggerOffset == null) {
    staggerOffset =
      windowStagger.getStaggerOffset(context.getCurrentProcessingTime(), size);
  }
  long start =
    TimeWindow.getWindowStartWithOffset(
    now, (globalOffset + staggerOffset) % size, size);
  return Collections.singletonList(new TimeWindow(start, start + size));
}

有三个参数element用户发送数据流数据，timestamp该数据流的时间戳，context窗口分配向下文信息，该方法在WindowOperator被调用。该类是处理事件的窗口类，所以该类不是不会使用timestamp该数据流的时间戳。由于windowStagger.getStaggerOffset(context.getCurrentProcessingTime(), size);返回0，所以staggerOffset就是0。

下面就是重点了，就是获取一个窗口的开始时间，获取开始时间后+窗口大小就是结束时间。

通过以上可知TimeWindow.getWindowStartWithOffset(now, (globalOffset + staggerOffset) % size, size);,第二个参数就是0，now是当前时间戳，size是窗口大小。

public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
  return timestamp - (timestamp - offset + windowSize) % windowSize;
}

上面的公式可以简化为timestamp - timestamp % windowSize;

也即是当前时间戳-时间戳在窗口多余的时间，肯定是窗口开始时间。

下面具体例子：2021-12-20 10:17:14 通过 2021-12-20 10:17:14 - 4秒，所以该数据落在[10:17:10-10:17:20)的窗口，同样在10:17:10-10:17:19产生的数据都会落在该窗口中。

所以实际上是不存在这些窗口的[10:17:11, 10:17:21)，[10:17:12, 10:17:22)，[10:17:14, 10:17:24)…

总结：窗口在定义时候，可以说窗口固化了窗口.
因为我们看到通过上面算法，每个时间都会落到特定的窗口。
后面会验证第一个窗口的开始时间是1970-01-01 08:00:00

再举一个例子：窗口大小改为8秒。

下面是一分钟内，各个处理时间对一个的窗口

用上面的时间验证第一个窗口的开始时间。例如上面第一个时间是2021-12-20 16:57:13那么也就是1639990633000，所以(1639990633000-0)/8=204998829 (取整），然后204998829 * 8 = 1639990632000 也就是2021-12-20 16:57:12就是这个窗口的开始时间。
ok 上面0就是1970-01-01 08:00:00

我们WindowOperator就是封窗用户WindowFunction处理功能的类,下面贴下代码

@Override
public void processElement(StreamRecord element) throws Exception {
 final Collection elementWindows =
   windowAssigner.assignWindows(
   element.getValue(), element.getTimestamp(), windowAssignerContext);
 	...
}

对于事件时间也是一样处理，只是不是使用当前时间计算所属的窗口，而是使用事件时间

@Override
public Collection assignWindows(
  Object element, long timestamp, WindowAssignerContext context) {
  if (timestamp > Long.MIN_VALUE) {
    if (staggerOffset == null) {
      staggerOffset =
        windowStagger.getStaggerOffset(context.getCurrentProcessingTime(), size);
    }
    // Long.MIN_VALUE is currently assigned when no timestamp is present
    long start =
      TimeWindow.getWindowStartWithOffset(
      timestamp, (globalOffset + staggerOffset) % size, size);
    return Collections.singletonList(new TimeWindow(start, start + size));
  } else {
    throw new RuntimeException(
      "Record has Long.MIN_VALUE timestamp (= no timestamp marker). "
      + "Is the time characteristic set to 'ProcessingTime', or did you forget to call "
      + "'DataStream.assignTimestampsAndWatermarks(...)'?");
  }
}