对流的联合进行排序以识别Apache Flink中的用户会话

Question

我有两个事件流

• L=（l1， l3， l8，...）-更稀疏，表示用户登录IP
• E=（e2， e4， e5， e9，...）-是特定IP的日志流

较低的索引表示时间戳...如果我们将两个流连接在一起并按时间对它们进行排序，我们将得到：

• l1， e2， l3， e4， e5， l8， e9，...

能否实现自定义的< code > Window /< code > Trigger 函数来将事件分组到会话(不同用户登录之间的时间):

< li>l1 - l3 : e2 < li>l3 - l8 : e4，e5 < li>l8 - l14 : e9，e10，e11，e12，e13 <李>...

我看到的问题是，这两个流不一定是排序的。我考虑过按时间戳对输入流进行排序。然后，使用全局窗口和自定义触发器可以很容易地实现窗口-但这似乎是不可能的。

在当前的Flink（v1.3.2）中，我是否遗漏了一些东西，或者绝对不可能这样做？

谢啦

Answer 1

问：E3不应该在L4之前吗？

使用< code>ProcessFunction排序非常简单。大概是这样的:

public static class SortFunction extends ProcessFunction<Event, Event> {
  private ValueState<PriorityQueue<Event>> queueState = null;

  @Override
  public void open(Configuration config) {
    ValueStateDescriptor<PriorityQueue<Event>> descriptor = new ValueStateDescriptor<>(
        // state name
        "sorted-events",
        // type information of state
        TypeInformation.of(new TypeHint<PriorityQueue<Event>>() {
        }));
    queueState = getRuntimeContext().getState(descriptor);
  }

  @Override
  public void processElement(Event event, Context context, Collector<Event> out) throws Exception {
    TimerService timerService = context.timerService();

    if (context.timestamp() > timerService.currentWatermark()) {
      PriorityQueue<Event> queue = queueState.value();
      if (queue == null) {
        queue = new PriorityQueue<>(10);
      }
      queue.add(event);
      queueState.update(queue);
      timerService.registerEventTimeTimer(event.timestamp);
    }
  }

  @Override
  public void onTimer(long timestamp, OnTimerContext context, Collector<Event> out) throws Exception {
    PriorityQueue<Event> queue = queueState.value();
    Long watermark = context.timerService().currentWatermark();
    Event head = queue.peek();
    while (head != null && head.timestamp <= watermark) {
      out.collect(head);
      queue.remove(head);
      head = queue.peek();
    }
  }
}

更新：有关通常更好的方法的描述，请参阅如何使用Flink对乱序事件时间流进行排序。