【发布时间】：2020-12-23 18:45:03
【问题描述】：
我看到有很多关于为每个密钥添加水印支持的讨论。但是 flink 是否支持每个分区的水印？
目前 - 然后考虑所有水印（非空闲分区）中的最小值。因此，窗口中最后一个挂起的记录也被卡住了。（使用periodicemit增加水印时）
非常感谢任何有关此的信息！
【问题讨论】：
标签： apache-flink
【发布时间】：2020-12-23 18:45:03
【问题描述】：
一些来源，例如FlinkKafkaConsumer，支持每个分区的水印。您可以通过在源上调用assignTimestampsAndWatermarks 来获得此信息，而不是在源生成的流上。
这样做是每个消费者实例跟踪每个分区内的最大时间戳，并将这些最大值中的最小值减去配置的有界无序作为其水印。空闲分区将被忽略，如果您将其配置为这样做。
这不仅会产生更准确的水印，而且如果您的事件在每个分区中是有序的，这也可以利用 WatermarkStrategy.forMonotonousTimestamps() 策略。
更多详情请见Watermark Strategies and the Kafka Connector。
至于为什么最后一个窗口没有被触发，这与水印有关，但与每个分区的水印无关。问题只是窗口是由水印触发的，而水印落后于事件中的时间戳。所以水印永远赶不上最后的事件，也永远无法触发最后一个窗口。
这对于无限制的流式作业来说不是问题，因为它们永远不会停止，也永远不会有最后一个窗口。对于批处理作业来说这不是问题，因为它们知道所有数据。但是对于有限的流式传输作业，您需要做一些事情来解决这个问题。从广义上讲，你必须做的是通知 Flink 输入流已经结束——每当 Flink 源检测到它们已经到达基于事件时间的输入流的末尾时，它们就会发出最后一个值为 MAX_WATERMARK 的水印，这将触发任何打开的窗口。
一种方法是使用KafkaDeserializationSchema 和isEndOfStream 的实现，当作业结束时返回true。
【讨论】：
-
谢谢大卫-我在源头使用水印（这里是kafka消费者级别）-即使水印保持在分区级别，当它们被发送给其他运营商时，flink实际上是在计算所有这些的分钟并将其发送出去。我通过在 StatusWatermarkValve.class 中调试得到它，当我在具有不同事件时间戳的多个分区中有数据时（因为我不能保证顺序） - 所有这些时间戳的最小值都发送给操作员。我在这里错过了什么吗？
-
我已经扩展了我的答案。
-
如果我有 3 个分区和 3 个任务管理器，每个 TM 内有 1 个任务槽分配给每个分区，那么会有 3 个水印（每个 TM/分区 1 个）或 1 个水印（最小所有这 3 个？）。我已经测试了这种情况，似乎只有一个水印（所有 3 个中的最小值）被发送到所有 TM。但是文档说，如果我们在源代码级别设置它们，我们可以为每个分区设置水印。我将在我的问题中添加更多细节。你能检查并解释发生了什么吗？
-
每个插槽都会做自己独立的水印。不同时隙或 TM 之间的水印没有全局协调。但是，任何具有多个输入的任务（例如，在 keyBy 之后）都将使用传入水印的最小值作为自己的水印。
-
谢谢 - 具体来说 - keyBy 运算符是如何工作的？它是否对流进行重新分区并将记录分配给其他一些任务管理器？在我的情况下 - 我确保具有相同键的记录被发送到同一个分区，但它们在第一个被消耗后仍然最终出现在其他一些不同的任务管理器中。这也可能是水印传播的原因。例如：如果我将 6 条记录（3 个差异键）发送到 3 个分区 - 它们被 3 个 TM 消耗，但在 keyBy 运算符之后最终在不同的 TM 中。