每 X 分钟发出一次会话 window 的结果

Question

我实现了一个 Flink 处理器，它将事件聚合到会话中，然后将它们写入接收器。现在我想扩展它，以便我可以每五分钟获取一次并发会话数。

进入我系统的事件的形式是：

{
  "SessionId": "UniqueUUID",
  "Customer": "CustomerA",
  "EventType": "EventTypeA",
  [ ... ]
}

并且单个会话通常包含多个不同 EventType 的事件。然后，我通过在 Flink 中执行以下操作将事件聚合到会话中。

DataStream<Session> sessions = events
                .keyBy((KeySelector<HashMap, String>) event -> (String) event.get(Field.SESSION_ID))
                .window(ProcessingTimeSessionWindows.withGap(org.apache.flink.streaming.api.windowing.time.Time.minutes(5)))
                .trigger(SessionProcessingTimeTrigger.create())
                .aggregate(new SessionAggregator())

当处理具有特定 EventType 的事件（“EventType”：“Session.Ended”）时，每个会话（由 SessionProcessingTimeTrigger）发出。最后将流发送到接收器并写入 Kafka。

现在我想写一个类似的 Flink 处理器，但不是只在会话完成后发出一个会话，而是想每 5 分钟发出一次所有会话，以便跟踪我们每 5 分钟有多少个并发会话分钟。所以从某种意义上说，我想我想要的是一个 SessionWindow，它也定期发出它的内容而不清除内容。

我对如何在 Flink 中完成此操作感到困惑，因此正在寻求帮助。

Answer 1

每当您希望 Flink window 在 non-default 次发出结果时，您可以通过实现自定义触发器来实现。每次 5-minute-long 计时器触发时，您的触发器只需要 return FIRE，除了它的原始逻辑。您需要在第一个事件分配给 window 时注册此计时器，并在每次计时器触发时再次注册。

在会话 windows 的情况下，由于合并会话 windows 的方式，这可能会更复杂。但我相信在处理时间会话的情况下 windows 我上面概述的内容会起作用。

每 X 分钟发出一次会话 window 的结果

Emitting the results of a session window every X minutes

apache-flink