Flink 中的会话窗口给出了意想不到的结果

Question

我有一个由两个字段“键入”的记录流，然后分配了一个间隔为 30 秒的会话窗口。我使用附加到记录的“时间戳”作为事件时间。我正在使用“assignAscendingTimestamps”水印。

以以下记录为例。流由 (user,place) 键入。

记录1：用户1，地点1，时间戳t1

Record2 : user2, place1, t1 后 30 秒的时间戳

Record3 : user1, place1, t1 30 秒内的时间戳

Record4 : user1, place1, t1 后 30 秒的时间戳

Record2 属于 user2，因此它属于不同的存储桶，因为流是键控的。因此，我希望 Record1、Record3 和 Record4 属于一个存储桶，而 Record2 属于另一个存储桶。

Bucket1

记录1：用户1，地点1，时间戳t1

Record3 : user1, place1, t1 30 秒内的时间戳

Record4 - user1, place1, t1 后 30 秒的时间戳

Bucket2

Record2 : user2, place1, t1 后 30 秒的时间戳

据我了解，只有 Record4 到达时才会触发包含 Record1 和 Record3 的会话窗口。但是当我运行代码时，一个只包含 Record1 的会话在 Record2 到达时被触发，因为 Record2 的时间戳在 Record1 的时间戳的时间间隔（30 秒）之后，尽管 Record2 的密钥不同。我浏览了 Flink 的文档和几个我可以在网上找到的 Session Windows 示例。然而我无法解决这个问题。我在这里缺少什么吗？这可能是因为上升的时间戳水印吗？

Answer 1

问题在于assignAscendingTimestamps 要求您的时间戳在所有键上单调递增。原因是 Flink 无法为每个 key 生成水印。

更新

由于 Flink 不能为每个键生成水印，因此必须生成水印，以便它们对所有元素都有效。如果每个键的时间戳是单调的，但不是跨所有键，那么您必须定义两个键之间的最大无序（时间戳差异）。通过从元素的时间戳中减去无序，您将获得有效的水印。另见BoundedOutOfOrdernessTimestampExtractor。但是请注意，如果元素以更大的无序到达，那么这也会中断。