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;
}

省略用于状态管理和一些记忆优化的代码，这似乎在流式用例中运行良好：在班次结束时间之后进入的第一个事件会触发最后班次的触发和聚合。

但是，作业可以在日期参数的范围内运行（例如：重新处理过去的时间段），或者由于一系列预期的原因而过早关闭。当这种事情发生时，我观察到最后一个窗口没有被触发/刷新，

ie：一天的最后一个班次在午夜结束，应该马上结束 开始第二天的第一班。活动在下午 23:59 开始， 轮班即将结束。然而，这项工作只是在 今天，00:00 结束。由于没有新元素到达 自定义触发器通过该行触发窗口触发， 不计算最后一个班次的聚合，但是，有些 部分结果仍然是预期的，即使没有发生任何事情 下一个班次或工作在正在进行的中间终止 换班。

我已经读到这样做的原因是：

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 以损害低级进程功能，他们通过将元素作为操作符的托管状态“手动”计算窗口，而在给定它们适合的规则或条件，并从定义的聚合函数或累加器中提取结果。同样在处理函数中，可以通过点击onClose 钩子来确定任何待处理的计算。

是否有一种方法可以通过扩展 Windows API 中的任何对象来获得每天特定时间的重复事件时间窗口的概念？

Answer 1

如果我理解正确，这里有两个单独的问题/问题需要解决：

1. 如何处理没有统一的窗口边界。
2. 如何在不丢失最后一个窗口的结果的情况下终止作业。

对于 (1)，您使用 GlobalWindows 和自定义 ShiftTrigger 的方法是一种方法。如果您想探索使用流程函数的替代方案，我已经编写了一个示例，您可以找到in the Flink docs。

要获得更流畅的 API，您可以创建自定义 WindowAssigner，然后可以利用内置的 EventTimeTrigger 作为其默认触发器。为此，您需要实现WindowAssigner 接口。

对于 (2)，只要您依赖事件时间处理，就不会触发最后一组窗口，除非在作业终止之前到达足以关闭它们的 Watermark。这通常要求您有一个事件，其时间戳在窗口结束之后足以创建一个足以触发窗口的水印（并且该作业保持运行足够长的时间以使其发生）。

但是，当 Flink 意识到一个流式作业即将自然结束时，它会自动注入一个时间戳设置为 MAX_WATERMARK 的 Watermark，其作用是触发所有事件时间计时器，并关闭所有事件时间窗口.对于任何有界源，这都会自动发生。使用 Kafka（例如），您还可以通过让您的反序列化器从 isEndOfStream 返回 true 来安排这一点。

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