对intervalJoin感到困惑

Question

我正在尝试提出一种解决方案，它涉及在连接操作之后应用一些逻辑，从多个事件b中的streamB中选择一个事件。这类似于reduce函数，但它只返回1个元素，而不是递增地返回。因此最终结果将是单个（EventA，EventB对，而不是一个EventA>/code>和多个EventB的叉积。

streamA
      .keyBy((a: EventA) => a.common_key)
      .intervalJoin(
          streamB
            .keyBy((b: EventB) => b.common_key)
        )
      .between(Time.days(-30), Time.days(0))
      .process(new MyJoinFunction)

数据将被摄取（假设它们具有相同的密钥）：

EventB ts：1616686386000

< code > EventB ts:1616686387000

事件 B ts： 1616686388000

EventB ts:1616686389000

事件 A ts： 1616686390000

每个EventA键保证只到达一次。

假设像上面这样的连接操作，它用4个事件b生成了1个EventA，成功地连接并收集在MyJoinFunction。现在，我想做的是，立即访问这些值，并执行一些逻辑以将事件a正确匹配到一个EventB。例如，对于上面的数据集，我需要（EventA1616686390000，EventB和1616686387000）。

将为每个(< code>EventA，< code>EventB)对调用< code>MyJoinFunction，但我希望在此之后有一个操作，它允许我访问一个迭代器，以便我可以查看每个< code>EventA的所有< code>EventB事件。

我知道我可以在连接后应用另一个窗口操作来对所有对进行分组，但我希望在连接成功后立即进行。因此，如果可能，我希望避免添加另一个窗口，因为我的窗口已经很大（30天）。

Flink是这个用例的正确选择，还是我完全错了？

Answer 1

这可以实现为KeyedCoProcessFunction。您将通过它们的公共密钥对两个流进行键值，连接它们，并一起处理两个流。

您可以使用ListState存储来自B的事件（对于给定键），以及ValueState存储来自a的事件（同样，对于给定键）。您可以使用事件时间计时器来了解何时可以查看ListState中的B事件，并生成结果。完成后不要忘记清除状态。

如果您不熟悉Flink API的这一部分，有关流程函数的教程应该会有所帮助。