如何解决如何测试WindowStore保留期?
我正在尝试对传入的kafka消息进行重复数据删除(我正在轮询一个数据源,该数据源使给定日期的第二天的所有数据点都在第二天可用,但时间不一致,所以我每隔x分钟轮询一次,我想对数据点进行重复数据删除,以得到一个仅包含新点的干净的下游主题。
为此,我建立了一个自定义转换器,该转换器依赖于商店来跟踪已处理的“点”。由于数据点的日期时间是重复数据删除密钥的一部分,因此我拥有一组无限制的密钥,因此我不能依赖简单的KeyValueStore。据我了解,WindowStore将允许我仅将密钥保留特定的保留期限(以我为例,为2天),所以这就是我所使用的。
我尝试使用 kafka-streams-test-utils 测试重复数据删除。重复数据删除效果很好,但是windowStore似乎无法“忘记”密钥。我尝试使用较短的窗口大小和持续时间(1s),但仍然无法忘记保留期后的键/值。
商店的配置::我希望对象在商店中停留2秒钟左右
config.put(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG,"1");
...
final StoreBuilder<WindowStore<String,AvroBicycleCount>> deduplicationStoreBuilder = Stores.windowStoreBuilder(
Stores.persistentWindowStore(deduplicationStore,Duration.ofSeconds(1),false),Serdes.String(),StreamUtils.AvroSerde()
);
我的变压器逻辑
@Override
public DataPoint transform(final String dataId,final DataPoint incoming) {
String key = dataId+"_"+incoming.getDateTime();
DataPoint previous = windowStore.fetch(key,incoming.getDateTime());
if(previous != null)
return null;
windowStore.put(key,incoming,incoming.getDateTime());
return incoming;
}
第三次测试失败
inputTopic.pipeInput("a",newDataPoint);
assertEquals(1,outputTopic.readRecordsToList().size(),"When a new data is emitted,it should go through");
inputTopic.pipeInput("a",newDataPoint);
assertEquals(0,"When the same data is re-emitted,it should not go through");
TimeUnit.SECONDS.sleep(10);
inputTopic.pipeInput("a","When the same data is re-emitted well past the retention period,it should go through");
关于windowStore的保留,我是否有一些不正确的理解?
解决方法
WindowedStore
在内部使用所谓的段来使数据过期。即,您的保留时间的时间范围被分为较小的时间范围,每个时间范围都有一个段来存储相应的数据(内部,一个段映射到一个商店,即{{1} }实际上是内部的多个商店)。如果段中的所有记录均已过期,则通过删除相应的存储来删除整个段(这比逐条记录的失效效率更高)。
此外,最小(硬编码)段大小为60秒,并且段数为2(硬编码),以避免太小(且效率低下)的段。因此,对于2天的保留时间,您将获得两个细分,每个细分的时间范围均为1天。因此,数据(在段的开头)最多可以保留3天,直到删除旧段为止。
因此,数据被有效地删除了一段时间。您无法配置细分数
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。