你好，我在Kafka会话窗口与非活动时间5分钟的工作。我需要一些反馈，当非活动时间达到和会话是关键下垂。假设我有（甲，1）记录“a”是键的位置。现在，如果我在5分钟内没有得到任何“a”键记录，会话将被丢弃。我想在会话结束时做一些操作，比如说（value）*2。有没有什么方法可以使用kafka流api来实现这一点
vs91vp4v1#
Kafka流不会在间隔时间过后丢弃会话。相反，如果另一个具有相同密钥的记录在间隔时间过后到达，if将创建一个新会话，并并行维护两个会话。这允许处理无序数据。甚至有可能发生这样的情况：如果一个无序的数据落入一个间隙中，并将两个会话相互“连接”，那么两个会话就会合并。默认情况下，会话保持1天。你可以通过 SessionWindows#until() 方法。如果会话过期，它将被无声地丢弃。没有通知。您还需要考虑配置参数 window.store.change.log.additional.retention.ms :默认保留设置为windows#maintainms（）+1天。您可以通过在streamsconfig中指定streamsconfig.window\u store\u change\u log\u additional\u retention\u ms\u config来覆盖此设置。因此，如果您想在时间流逝时做出React，您应该查看允许您注册基于“偶数时间进度”或挂钟时间的常规回调（某种计时器）的标点符号。如果某个会话在一段时间内没有更新，并且您认为该会话“已完成”，那么您就可以做出React。
SessionWindows#until()
window.store.change.log.additional.retention.ms
1条答案
按热度按时间vs91vp4v1#
Kafka流不会在间隔时间过后丢弃会话。相反，如果另一个具有相同密钥的记录在间隔时间过后到达，if将创建一个新会话，并并行维护两个会话。这允许处理无序数据。甚至有可能发生这样的情况：如果一个无序的数据落入一个间隙中，并将两个会话相互“连接”，那么两个会话就会合并。
默认情况下，会话保持1天。你可以通过
SessionWindows#until()方法。如果会话过期，它将被无声地丢弃。没有通知。您还需要考虑配置参数window.store.change.log.additional.retention.ms:默认保留设置为windows#maintainms（）+1天。您可以通过在streamsconfig中指定streamsconfig.window\u store\u change\u log\u additional\u retention\u ms\u config来覆盖此设置。
因此，如果您想在时间流逝时做出React，您应该查看允许您注册基于“偶数时间进度”或挂钟时间的常规回调（某种计时器）的标点符号。如果某个会话在一段时间内没有更新，并且您认为该会话“已完成”，那么您就可以做出React。