Apache Flink，每日轮班的事件时间窗口

Question

我的Flink工作必须在每次工作轮班后计算某个集合。换挡是可配置的，看起来类似于：

1st shift: 00:00am - 06:00am
2nd shift: 06:00am - 12:00pm
3rd shift: 12:00pm - 18:00pm

出于操作目的，每天的班次都是一样的，一周/一年中的几天之间没有区别。轮班配置可以随时间变化，并且可以不单调，因此表中留下了一个简单的EventTime窗口，如:TumblingEventTimeWindows.of(time.of(6,HOURS))，因为一些轮班可能会缩小或超时，或者在中间插入几个小时...

我想出了一些基于GlobalWindow和自定义触发器的东西：

LinkedList<Shift> shifts;

datastream.windowAll(GlobalWindows.create())
          .trigger(ShiftTrigger.create(shifts))
          .aggregate(myAggregateFunction)

在我的自定义触发器中，我尝试识别传入事件是否超过了正在进行的工作轮班的结束时间，并触发该轮班的窗口：

@Override
public TriggerResult onElement(T element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {
    // compute the end time of the on-going shift
    final Instant currentShiftEnd = ...
    // fire window for the shift if the event passes the end line
    if (ShiftPredicate.of(currentShiftEnd).test(element)) {
        return TriggerResult.FIRE_AND_PURGE;
    }
    return TriggerResult.CONTINUE;
}

省略了状态管理和一些记忆优化的代码，这似乎在流式用例中工作得很好：shift endtime之后进入的第一个事件触发了最后一个shift的触发和聚合。

我读到这样做的原因是：

Flink只保证基于时间的窗口的删除，而不保证其他类型的删除，例如全局窗口（请参阅窗口分配器）

我已经在org.apache.flink.streaming.api.windowing包中查看了一些类似TumblingEventTimeWindows或DynamicEventTimeSessionWindows的东西，可以使用这些东西，也可以使用一天中的一个结束小时进行扩展，这样，当元素的水印超过窗口限制时，我就可以依赖这些触发的默认事件时触发器，但我不确定如何做到这一点。凭直觉我会希望这样的事情：

shifts.forEach(shift -> {
    datastream.windowAll(EventTimeWindow.fromTo(DAILY, shift.startTime, shift.endTime))
              .aggregate(myAggregateFunction);
});

我知道对于任意复杂度的用例，有些人所做的是抛弃Windows API，以损害低级进程函数，在低级进程函数中，他们通过将元素保持为运算符的受管状态来“手动”计算窗口，而在给定的规则或条件下，他们拟合并从定义的聚合函数或累加器中提取结果。在process函数中，也可以通过点击onclose钩子来pin point任何挂起的计算。

是否有一种方法可以通过扩展windows API中的任何对象来获得一天中某几个小时的循环事件时间窗口的概念？

Answer 1

如果我没有理解错的话，这里有两个独立的问题/问题需要解决：

1. 如何处理不具有统一的窗口边界。
2. 如何终止作业而不丢失上一个窗口的结果。

对于(1)，您使用GlobalWindows和自定义ShiftTrigger的方法是一种可行的方法。如果您想探索使用process函数的替代方法，我编写了一个示例，您可以在Flink文档中找到。

另一种处理方法是避免在这些作业完成时取消它们，而是使用./bin/flink stop--drain[-p savepointPath] 干净地停止作业（使用保存点），同时排出所有剩余的窗口结果（通过注入最后一个大水印(MAX_WATERMARK))。