如何在窗口和接收器操作符之间注入延迟？

Question

上下文 - 应用程序

我们有一个处理事件的 Apache Flink 应用程序

• 应用程序使用事件时间特征
• 应用程序分片 (keyBy) 事件基于 sessionId 字段
• 应用程序的窗口化为 1分钟翻滚窗口
  • 窗口由 reduce 和 process 函数指定
  • 因此，对于每个会话，我们将有 1 条计算记录
• 应用程序将数据发送到 Postgres 接收器

上下文 - 基础设施

应用程序中：

• 它通过 Kinesis Data Analytics (KDA) 在 AWS 中托管
• 它在 5 个不同的区域中运行
• 每个区域中运行完全相同的代码

数据库：

• 它通过 RDS 在 AWS 中托管（目前它是 PostgreSQL）
• 它位于一个区域（带有只读副本在不同的区域）

问题

因为我们使用具有 1 分钟滚动窗口的事件时间特征，所以所有区域的接收器几乎同时发出其记录。

我们想要实现的是在窗口和接收器操作符之间添加人为延迟以推迟接收器发射。

Flink 应用程序	偏移	窗口 1	接收器第一次运行	窗口 2	接收器第二次运行
#1	0	60	60	120	120
#2	12	60	72	120	132
#3	24	60	84	120	144
#4	36	60	96	120	156
#5	48	60	108	120	168

不起作用的解决方法

我们认为可以为 驱逐者的 evictBefore 像这样

...
.keyBy(event -> event.getSessionId())
.window(getWindowAssigner(config))
.allowedLateness(Time.seconds(config.getWindowLatenessInSec()))
.evictor(new Evictor<>() {
    private static final long serialVersionUID = 5373966807521260856L;

    public void evictBefore(Iterable<TimestampedValue<Event>> iterable, int i, TimeWindow timeWindow, EvictorContext evictorContext) {
        try {
            Thread.sleep(config.getWindowingDelayInMilliSec());
        } catch (InterruptedException ignore) {
        }
    }

    @Override
    public void evictAfter(Iterable<TimestampedValue<Event>> iterable, int i, TimeWindow timeWindow, EvictorContext evictorContext) {

    }
})
...

，但它不能可靠地工作。

Answer 1

您可以将 TumblingEventTimeWindows of(Time size, Time offset, WindowStagger windowStagger) 与 WindowStagger.RANDOM 结合使用。

请参阅 https://nightlies.apache.org/flink/flink-docs-stable/api/java/org/apache/flink/streaming/api/windowing/assigners/WindowStagger.html用于文档。