让我们考虑一个主题，它有多个分区，消息按事件时间顺序编写，没有任何特定的分区方案。kafka streams应用程序对这些消息进行一些转换，然后按某个键分组，然后按给定宽限期的事件时间窗口聚合消息。
每个任务可以以不同的速度处理传入消息（例如，因为在具有不同性能特征的服务器上运行）。这意味着在groupby shuffle之后，当内部主题的同一分区中的消息来自不同的任务时，它们之间的事件时间顺序将不会保留。一段时间后，此事件时间偏差可能会变得比宽限期更大，这将导致删除来自滞后任务的消息。
增加宽限期似乎不是一个有效的选项，因为它会延迟发出最终的聚合结果。apacheflink通过在合并分区上发出最低的水印来处理这个问题。
它应该是一个真正的问题，特别是在处理大量历史数据时，还是我遗漏了什么？Kafka流是否提供了一个解决方案来处理这种情况？
更新我的问题不是关于kstream kstream连接，而是关于在流洗牌之前的单个kstream事件时间聚合。
考虑以下代码段：
stream
.mapValues(...)
.groupBy(...)
.windowedBy(TimeWindows.of(Duration.ofSeconds(60)).grace(Duration.ofSeconds(10)))
.aggregate(...)我假设mapvalues（）操作对于某些任务来说可能会很慢，不管是什么原因，因为这些任务确实以不同的速度处理消息。当洗牌发生在 aggregate() 运算符，任务0可能一直在处理消息 t 当任务1还在运行时 t-skew ，但来自这两个任务的消息最终交错在内部主题的单个分区中（对应于分组键）。
我担心的是，当skew足够大（在我的示例中超过10秒）时，来自滞后任务1的消息将被丢弃。
1条答案
按热度按时间6xfqseft1#
基本上，任务/处理器维护一个流时间，它被定义为任何已经轮询的记录的最高时间戳。这个流时间然后在kafka流中用于不同的目的（例如：punctator、加窗聚合等）。
[窗口聚合]
如您所述，流时间用于确定是否应接受记录，即记录\u accepted=
end_window_time(current record) + grace_period > observed stream_time.如您所述，如果多个任务并行运行以基于分组键洗牌消息，并且某些任务比其他任务慢（或者某些分区脱机），则会创建无序消息。不幸的是，我恐怕唯一的办法就是增加
grace_period.这实际上是可用性和一致性之间永恒的权衡。
[kafkastream和kafkastream/ktable连接的行为]
当您使用kafka流执行连接操作时，一个内部任务被分配给多个共同分区的主题上的“同一”分区。例如，任务0将被分配给topic1-partition0和topicb-partition0。
获取的记录按分区缓冲到由任务管理的内部队列中。因此，每个队列包含单个分区等待处理的所有记录。
然后，从队列中逐个轮询记录，并由拓扑示例进行处理。但是，这是来自非空队列的记录，该队列具有从轮询队列返回的最低时间戳。
此外，如果队列为空，则任务可能在一段时间内处于空闲状态，这样就不会从队列中轮询更多的记录。实际上，您可以通过stream config来配置任务保持空闲的最长时间：
max.task.idle.ms这种机制允许同步共同本地化的分区。bu，默认max.task.idle.ms设置为0。这意味着任务永远不会等待来自分区的更多数据，这可能会导致记录被过滤，因为流时间可能会更快地增加。