如何在ApacheFlink中为会话窗口分配id？

Question

如何在ApacheFlink中为会话窗口分配id？

最后，我希望在会话窗口打开时，使用会话窗口id逐个充实事件（我不希望等到窗口关闭后再发出充实事件）。

我尝试使用AggregateFunction来实现这一点，但是我认为merge（）并没有像我所期望的那样工作。它似乎是用于合并窗口而不是窗格（触发触发）。在我的管道中似乎从未调用过它。因此，触发器之间似乎没有共享状态！

会话窗口ID将是落入窗口的第一个事件的时间戳（由于非保证排序，这可能意味着一些事件可能会落入具有较早时间戳的同一会话窗口-我对此没意见）。

public class FooSessionState {

  private Long sessionCreationTime;

  private FooMatch lastMatch;
}

/**
 * Aggregator that assigns session ids to elements of a session window
 */
public class SessionIdAssigner implements
    AggregateFunction<FooMatch, FooSessionState, FooSessionEvent> {

  static final long serialVersionUID = 0L;

  @Override
  public FooSessionState createAccumulator() {
    return new FooSessionState();
  }

  @Override
  public FooSessionState add(FooMatch value, FooSessionState sessionState) {
    if (sessionState.getSessionCreationTime() == null) {
      sessionState.setSessionCreationTime(value.getReport().getTimestamp());
    }
    sessionState.setLastMatch(value);
    return sessionState;
  }

  @Override
  public FooSessionEvent getResult(FooSessionState accumulator) {
    FooSessionEvent sessionEvent = new FooSessionEvent();
    sessionEvent.setFooMatch(accumulator.getLastMatch());
    sessionEvent.setSessionCreationTime(accumulator.getSessionCreationTime());
    return sessionEvent;

  }

  @Override
  public FooSessionState merge(FooSessionState a, FooSessionState b) {

    if ( a.getSessionCreationTime() != null) {
      b.setSessionCreationTime(a.getSessionCreationTime());
    }

    return b;
  }
}

我的计划是按如下方式使用它：

stream.keyBy(new FooMatchKeySelector())
    .window(EventTimeSessionWindows.withGap(Time.milliseconds(config.getFooSessionWindowTimeout())))
    .trigger(PurgingTrigger.of(CountTrigger.of(1L)))
    .aggregate(new SessionIdAssigner())

Answer 1

我认为会话窗口不适合您想要实现的目标。它们被设计为每个会话聚合事件，但不是丰富每个事件，也就是说，它们计算结果并在窗口关闭时发出结果。正如您所注意到的，会话窗口通过为每个事件创建一个新窗口并合并重叠的窗口来工作。之所以选择这种设计，是因为事件可能会无序到达。因此，可能会出现这样的情况：您有两个窗口，稍后通过桥接事件连接。

我建议使用一个收集事件并对其时间戳进行排序的ProcessFunction来实现逻辑。当接收到水印时，它会发出带有正确会话ID的所有收集到的事件。因此，您只保留两个水印之间的事件状态。除了这些事件之外，您还需要保留最后发出的事件的时间戳和最后发出的会话ID来执行正确的会话化。