apache-flink - Flink Windows边界，水印，事件时间戳和处理时间

Question

问题定义与建立概念

假设我们有一个 TumblingEventTimeWindow ，其大小为5分钟。我们的事件包含2条基本信息:

号

事件时间戳

在此示例中，我们在 12:00 PM工作者计算机的挂钟时间(当然，工作者可以有不同步的时钟，但这不在此问题的范围内)启动 Flink 拓扑。该拓扑包含 和一个处理操作员，负责将的值归纳到中，该值属于每个窗口以及与该问题无关的KAFKA接收器。

此窗口有一个BoundedOutOfOrdernessTimestampExtractor，允许的延迟时间为一分钟。

水印:据我了解，Flink和Spark结构化流中的水印定义为( max-event-timestamp-seen-so-far-允许延迟)。事件时间戳等于或小于等于该水印的事件时间戳的任何事件都将被丢弃并在结果计算中忽略。

第1部分(确定窗口的边界)

快乐(实时)路径

在这种情况下，几个事件到达 Flink运算符，它们具有跨越 12:01 - 12:09的不同事件时间戳。此外，事件时间戳与我们的 处理时间(在下面的X轴中显示)相对一致。由于我们正在处理 EVENT_TIME 特性，因此应通过其 事件时间戳确定事件是否属于某个特定事件。

enter image description here

旧数据涌入

在该流程中，我让 假定，这两个 滚动窗口的 边界分别是 12:00 -- 12:05和 12:05 -- 12:10，这是因为我们从 12:00 开始了拓扑的执行。如果该假设是正确的(我希望不是)，那么在回填情况下会发生什么情况，在这种情况下，随着 出现了多个旧的事件时间戳，而时间戳更久的事件时间戳记，我们在 12:00开始了拓扑？ (年龄足够大，以致我们的迟到津贴不包括这些费用)。类似于以下内容:

enter image description here

如果是这样，那么我们的事件当然不会在任何窗口中捕获，因此，我再次希望不是这种行为:)

另一个选择是通过到达事件的事件时间戳确定窗口的边界。如果是这样，那将如何工作？注意到的最小事件时间戳成为第一个窗口的开始，并根据大小从该窗口开始(在这种情况下为 5分钟)，确定了随后的边界吗？因为这种方法也会有缺陷和漏洞。您能否解释一下这是如何工作的以及如何确定窗口的边界？

回填事件涌入

上一个问题的答案也将解决此问题，但是我认为在此处明确提及它会有所帮助。假设我有 大小为5分钟的 TumblingEventTimeWindow 。然后在 12:00 ，我启动了一项回填工作，该工作在许多事件中涌向Flink运算符，该运算符的时间戳覆盖 10:02 - 10:59范围；但是，由于这是一项回填工作，因此 整个执行大约需要 3分钟才能完成。

作业是否将分配 12个单独的窗口，并根据事件的事件时间戳正确地填充它们？那 12个窗口的边界是什么？我是否将以 结束12个输出事件，每个输出事件的 总结了每个分配窗口的值？

第2部分(此类有状态运营商的单元/集成测试)

对于此类逻辑和运算符的自动化测试，我也有一些担忧。操纵 处理时间的最佳方法，以某种方式触发某些行为，以便塑造所需窗口的边界以进行测试。特别是因为到目前为止，我所读到的关于使用 Test Harnesses的内容似乎有些混乱，并且可能会导致一些混乱的代码，而这些代码并不那么容易阅读:

Unit Test Stateful Operators

Lateness Testing of Window in Flink

参考文献

我在这方面学到的大部分知识和一些困惑的根源可以在以下地方找到:

Timestmap Extractors & Watermark Emitters

Event Time Processing & Watermarking

Handling Late Data & Watermarking in Spark

Spark文档该部分中的图像 super 有用且富有教育意义。但是同时，窗口边界与处理时间和而不是事件时间戳对齐的方式给我造成了一些困惑。

此外，在该可视化中，似乎水印每 5分钟就会计算一次，因为这是窗口的滑动规范。这是应该多久计算一次水印的决定因素吗？在 Flink 中针对不同的窗口(例如Tumbling，Sliding，Session等)如何工作？

衷心感谢您的帮助，如果您对这些概念及其内部运作有更好的参考，请告诉我。

@snntrable回答后的更新

如果您使用事件时间语义来运行Job，则在窗口运算符上的处理时间是完全不相关的

没错，我理解这一部分。一旦处理了 EVENT_TIME特性，您就在语义/逻辑上与 处理时间脱节了。我推迟 处理时间的原因是我对以下关键问题感到困惑，这仍然是一个谜:

如何计算 窗口的边界 ？!

另外，非常感谢您澄清 out-of-orderness和 lateness之间的区别。我正在处理的代码因名称不正确而完全使我失望(从 BoundedOutOfOrdernessTimestampExtractor继承的类的构造函数参数称为 maxLatency):/

考虑到这一点，让我看看我是否可以正确理解 ，如何在时将水印计算为和 ，将事件丢弃(或侧面输出):

乱序分配器

当前水印= max-event-time-seen-so-far - max-out-of-orderness-allowed

允许的延迟

当前水印= max-event-time-seen-so-far - allowed-lateness

常规流

当前水印= max-event-time-seen-so-far

并且在中的任何一种情况中，无论事件时间戳时间戳小于或等于等于 current-watermark的事件，将 丢弃(侧输出)，对吗？

这就提出了一个新问题。您何时想使用 out of orderness而不是 lateness？由于 当前水印计算(在数学上)在这些情况下可以相同。当您同时使用 和时，会发生什么(甚至有意义)？

返回Windows的边界

这仍然是我的主要谜团。鉴于以上所有讨论，让我们回顾我提供的具体示例，看看如何在此处确定 窗口的边界。假设我们有以下情况(事件的形式为 (value, timestamp)):

操作员于 12:00 PM (即处理时间)开始

事件按以下顺序到达操作员

(1， 8:29 )

(5， 8:26 )

(3， 9:48 )

(7， 9:46 )

我们有一个 TumblingEventTimeWindow ，大小为 5分钟

该窗口将应用于具有 2分钟 DataStream

的 BoundedOutOfOrdernessTimestampExtractor的 maxOutOfOrderness

也是，该窗口还配置了的allowedLateness 1分钟

注意:如果您不能同时使用 out of orderness和 lateness或 不能使用有意义，请仅在中使用来考虑上述示例中的 out of orderness。

最后，您能否布置将为其分配一些事件的 窗口， 和，请指定这些窗口的 边界( 以开头， 结束，窗口的时间戳)。我假设边界是由 事件的时间戳以及决定的，但是在像这样的具体示例中弄清楚它们是有些棘手的。

再次，巨大的感谢，并衷心感谢您的帮助:)

Answer 1

原始答案

水印:据我所知，Flink和Spark结构化流中的水印定义为(最大事件时间戳-迄今可见-允许延迟)。事件时间戳小于或等于此水印的任何事件都将被丢弃并在结果计算中忽略。

这是不正确的，可能是造成混淆的原因。乱序和延迟是Flink中的不同概念。对于BoundedOutOfOrdernessTimestampExtractor，水印为max-event-timestamp-seen-so-far - max-out-of-orderness。 Flink文档[1]中有关允许延迟的更多信息。
如果您使用事件时间语义来运行Job，则在窗口运算符上的处理时间是完全不相关的:

事件将根据事件时间时间戳

分配给其窗口
一旦水印达到其最大时间戳( window end time -1)，就会触发

时间窗口。

时间戳早于 current watermark - allowed lateness的

事件将被丢弃或发送到后期数据侧输出[1]

这意味着，如果您在12:00 pm(处理时间)开始工作并开始提取过去的数据，则水印也会(甚至更早)出现。因此，配置的 allowedLateness是无关紧要的，因为数据相对于偶数时间不晚。
另一方面，如果您首先从12:00 pm提取一些数据，然后从10:00 pm提取数据，则在提取旧数据之前，水印已经提前到〜12:00pm。在这种情况下，晚上10:00起的数据将是“晚期”。如果它晚于已配置的 allowedLateness(默认= 0)，则将其丢弃(默认)或发送到侧面输出(如果已配置)[1]。
跟进答案
事件时间窗口的时间轴如下:

窗口中带有时间戳的第一个元素到达->创建此窗口的状态(＆键)

watermark >= window_endtime - 1到达->触发窗口(发出结果)，但状态不被丢弃

watermark >= window_endtime + allowed_latenes到达->状态被丢弃

在2和3.之间，此窗口的事件延迟，但在允许的延迟范围内。这些事件将添加到现有状态，并且-默认情况下-会在每个记录上触发该窗口，以发出精确的结果。
3.之后，该窗口的事件将被丢弃(或发送到后期的输出接收器)。
因此，是的，同时配置两者都是有意义的。乱序确定了第一次触发窗口的时间，而允许的延迟确定了将状态保留多长时间以可能更新结果。
关于边界:翻转事件时间窗口具有固定的长度，在各个键之间对齐，并从unix时代开始。空窗口，不存在。对于您的示例，这意味着:

(1，8:29)已添加到窗口(8:25-8:29:59:999)

(5，8:26)已添加到窗口(8:25-8:29:59:999)

(3，9:48)已添加到窗口(9:45-9:49:59:999)

(8:25-8:29:59:999)被触发，因为水印已前进到9:48-0:02 = 9:46，该值大于窗口的最后时间戳。窗口状态也会被丢弃，因为水印已前进到9:46，这也大于窗口的结束时间+允许的延迟时间(1分钟)

(7，9:46)添加到窗口中添加到窗口(9:45-9:49:59:999)

希望这可以帮助。
康斯坦丁
[1] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/operators/windows.html#allowed-lateness