如何测试windowstore保留期?

bvjveswy  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(240)

我正在尝试对传入的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的保留,是否有一些我不太理解的地方?

voase2hg

voase2hg1#

WindowedStore 在内部使用所谓的段使数据过期。例如,保留时间的时间范围被划分为更小的时间范围,每个时间范围都有一个段来存储相应的数据(在内部,一个段Map到一个存储,即 WindowedStore 实际上是内部的多个商店)。如果一个段中的所有记录都已过期,则通过删除相应的存储来删除整个段(这比逐个记录过期更有效)。
此外,最小(硬编码)段大小为60秒,段数为2(硬编码),以避免段太小(且效率低下)。因此,对于保留时间为2天的情况,可以得到两个段,每个段的时间范围为1天。因此,数据(在一个段的开始处)最长可保留3天,直到删除一个旧段为止。
因此,数据在一定的延迟下被有效地删除。不能配置段数

相关问题