Kafka流-时间窗口

Question

我有一个关于Kafka流的时间窗的问题，有些概念真的让我困惑。

我们有一个主题每天获得1000万个事件，我们有6天的日志保留，所以总的主题包含6000万个事件。

现在，我创建了一个KTable，我正在执行load all操作并迭代事件。正如我之前提到的，实际上我们只是当前的事件，而不是6千万事件，所以我在KTable定义中窗口化了这些数据。

.windowedBy(TimeWindows.of(TimeUnit.DAYS.toMillis(1)).until(TimeUnit.DAYS.toMillis(1))

现在，当我用以下语句加载所有事件时，一切都运行良好。

store().fetchAll(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1), System.currentTimeMillis())

问题是在一天的早些时候，这将加载100万个事件，但稍后将加载1000万个事件，所以我必须迭代超过1000万个事件，而我们是在批处理模式下工作，我想我可以进一步优化这一点，只加载事件最后一个小时，所以对于相同的KTable配置，我尝试使用以下语句。

store().fetchAll(System.currentTimeMillis() - TimeUnit.HOURS.toMillis(1), System.currentTimeMillis())

但令我惊讶的是，它没有返回任何数据。

有人能解释一下为什么这没有返回任何结果吗，我想我误解了时间窗口的概念。

然后我做了一些进一步的测试，并将我的KTable配置更改为followed。

.windowedBy(TimeWindows.of(TimeUnit.HOURS.toMillis(1)).until(TimeUnit.DAYS.toMillis(1)))

store().fetchAll(System.currentTimeMillis() - TimeUnit.HOURS.toMillis(1), System.currentTimeMillis())

但我不确定我走的是正确的路...

如果我将使用下面的语句来进行最新的KTable配置，那么这会从当前的一天向我传递1000万个事件吗？

store().fetchAll(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1), System.currentTimeMillis())

Answer 1

在窗口存储上使用交互式查询时，时间范围将应用于窗口开始时间戳。因此，如果您有一个1天的窗口，并从[现在-1小时，现在）查询具有窗口开始时间戳的数据，您将找不到任何匹配的窗口，因为在此时间范围内没有窗口开始。