我正在尝试对传入的kafka消息进行重复数据消除(我正在轮询一个数据源,该数据源使某一天的所有数据点在第二天都可用,但时间不一致,因此我每x分钟轮询一次,我希望对数据点进行重复数据消除,以便有一个只包含新点的干净的下游主题)。
为此,我构建了一个定制的转换器,它依赖于一个存储来跟踪哪个“点”已经被处理。由于数据点的datetime是重复数据消除密钥的一部分,因此我有一组无限的密钥,因此我不能依赖简单的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), 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, outputTopic.readRecordsToList().size(), "When the same data is re-emitted, it should not go through");
TimeUnit.SECONDS.sleep(10);
inputTopic.pipeInput("a", newDataPoint);
assertEquals(1, outputTopic.readRecordsToList().size(), "When the same data is re-emitted well past the retention period, it should go through");
关于WindowsStore的保留,是否有一些我不太理解的地方?
1条答案
按热度按时间voase2hg1#
一
WindowedStore
在内部使用所谓的段使数据过期。例如,保留时间的时间范围被划分为更小的时间范围,每个时间范围都有一个段来存储相应的数据(在内部,一个段Map到一个存储,即WindowedStore
实际上是内部的多个商店)。如果一个段中的所有记录都已过期,则通过删除相应的存储来删除整个段(这比逐个记录过期更有效)。此外,最小(硬编码)段大小为60秒,段数为2(硬编码),以避免段太小(且效率低下)。因此,对于保留时间为2天的情况,可以得到两个段,每个段的时间范围为1天。因此,数据(在一个段的开始处)最长可保留3天,直到删除一个旧段为止。
因此,数据在一定的延迟下被有效地删除。不能配置段数