为什么 Flink 在 DataStream join + Global 窗口上发出重复记录？

Question

我正在学习/体验 Flink，我观察到 DataStream 连接的一些意外行为，并想了解正在发生的事情......

假设我有两个流，每个流有 10 条记录，我想加入 id 字段。让我们假设一个流中的每条记录在另一个流中都有一个匹配的记录，并且每个流中的 ID 都是唯一的。假设我必须使用全局窗口（要求）。

使用 DataStream API 加入（我在 Scala 中的简化代码）：

val stream1 = ... // from a Kafka topic on my local machine (I tried with and without .keyBy)
val stream2 = ... 

stream1
  .join(stream2)
  .where(_.id).equalTo(_.id)
  .window(GlobalWindows.create()) // assume this is a requirement
  .trigger(CountTrigger.of(1))
  .apply {
    (row1, row2) => // ... 
  }
  .print()

结果：

• 一切都按预期打印，第一个流中的每条记录都与第二个流中的一条记录相结合。

但是：

• 如果我将其中一条记录（例如，带有更新的字段）从一个流重新发送到该流，则会发出两个重复的连接事件????
• 如果我重复该操作（有或没有更新的字段），我将得到 3 个已发出的事件，然后是 4、5 等等...... ????

Flink 社区中有人能解释一下为什么会这样吗？我本来预计每次只会发出 1 个事件。是否可以通过全局窗口来实现这一点？

相比之下，Flink Table API 在同一场景中的表现与预期一致，但对于我的项目，我对 DataStream API 更感兴趣。

Table API 示例，按预期工作：

tableEnv
  .sqlQuery(
    """
      |SELECT *
      |  FROM stream1
      |       JOIN stream2
      |       ON stream1.id = stream2.id
    """.stripMargin)
  .toRetractStream[Row]
  .filter(_._1) // just keep the inserts
  .map(...)
  .print() // works as expected, after re-sending updated records

谢谢，

尼古拉斯

Answer 1

问题是记录永远不会从您的全局窗口中删除。因此，只要有新记录到达，但旧记录仍然存在，您就会在全局窗口上触发连接操作。

因此，要让它在您的情况下运行，您需要实现自定义evictor。我在一个最小的工作示例中扩展了您的示例并添加了驱逐器，我将在 sn-p 之后解释。

val data1 = List(
  (1L, "myId-1"),
  (2L, "myId-2"),
  (5L, "myId-1"),
  (9L, "myId-1"))

val data2 = List(
  (3L, "myId-1", "myValue-A"))

val stream1 = env.fromCollection(data1)
val stream2 = env.fromCollection(data2)

stream1.join(stream2)
  .where(_._2).equalTo(_._2)
  .window(GlobalWindows.create()) // assume this is a requirement
  .trigger(CountTrigger.of(1))
  .evictor(new Evictor[CoGroupedStreams.TaggedUnion[(Long, String), (Long, String, String)], GlobalWindow](){
    override def evictBefore(elements: lang.Iterable[TimestampedValue[CoGroupedStreams.TaggedUnion[(Long, String), (Long, String, String)]]], size: Int, window: GlobalWindow, evictorContext: Evictor.EvictorContext): Unit = {}

    override def evictAfter(elements: lang.Iterable[TimestampedValue[CoGroupedStreams.TaggedUnion[(Long, String), (Long, String, String)]]], size: Int, window: GlobalWindow, evictorContext: Evictor.EvictorContext): Unit = {
      import scala.collection.JavaConverters._
      val lastInputTwoIndex = elements.asScala.zipWithIndex.filter(e => e._1.getValue.isTwo).lastOption.map(_._2).getOrElse(-1)
      if (lastInputTwoIndex == -1) {
        println("Waiting for the lookup value before evicting")
        return
      }
      val iterator = elements.iterator()
      for (index <- 0 until size) {
        val cur = iterator.next()
        if (index != lastInputTwoIndex) {
          println(s"evicting ${cur.getValue.getOne}/${cur.getValue.getTwo}")
          iterator.remove()
        }
      }
    }
  })
  .apply((r, l) => (r, l))
  .print()

在应用窗口函数（本例中为连接）后，将应用驱逐器。如果您在第二个输入中有多个条目，则尚不完全清楚您的用例应该如何工作，但目前，驱逐器仅适用于单个条目。

每当一个新元素进入窗口时，窗口函数立即被触发（count = 1）。然后使用具有相同键的所有元素评估连接。之后，为了避免重复输出，我们从当前窗口的第一个输入中删除所有条目。由于第二个输入可能在第一个输入之后到达，因此当第二个输入为空时不执行驱逐。请注意，我的 scala 生锈了；您将能够以更好的方式编写它。运行的输出是：

Waiting for the lookup value before evicting
Waiting for the lookup value before evicting
Waiting for the lookup value before evicting
Waiting for the lookup value before evicting
4> ((1,myId-1),(3,myId-1,myValue-A))
4> ((5,myId-1),(3,myId-1,myValue-A))
4> ((9,myId-1),(3,myId-1,myValue-A))
evicting (1,myId-1)/null
evicting (5,myId-1)/null
evicting (9,myId-1)/null

最后一句话：如果 table API 已经提供了一种简洁的方式来做你想做的事情，我会坚持下去，然后在需要时convert it to a DataStream。