kafka 流跳跃窗口聚合导致时间戳为零的多个窗口

Question

Kafka Streams DSL 窗口聚合导致多个窗口。

@StreamListener("input")
    public void process(KStream<String, Data> DataKStream) {

        JsonSerde<DataAggregator> DataJsonSerde =
                new JsonSerde<>(DataAggregator.class);

        DataKStream
                .groupByKey()
                .windowedBy(TimeWindows.of(60000).advanceBy(30000))
                .aggregate(
                        DataAggregator::new,
                        (key, Data, aggregator) -> aggregator.add(Data),
                        Materialized.with(Serdes.String(), DataJsonSerde)
                );
    }

DataAggregator.java

public class DataAggregator {

    private List<String> dataList = new ArrayList<>();

    public DataAggregator add(Data data) {
        dataList.add(data.getId());
        System.out.println(dataList);
        return this;
    }

    public List<String> getDataList() {
        return dataList;
    }
}

我根据键对输入数据进行分组，然后执行 1 分钟窗口和 30 秒跳跃，在聚合器中我只是收集数据并显示。

我希望在开始时有 1 个窗口，30 秒后会出现另一个窗口。但是实际的输出是不同的，因为开始它本身就创建了 2 个窗口。

预期：

[1]
[1, 2]
[1, 2, 3]
[1, 2, 3, 4]
[1, 2, 3, 4, 5]
[1, 2, 3, 4, 5, 6] // till 30 seconds only one window
[6] // new window after 30 seconds
[1, 2, 3, 4, 5, 6, 7]
[6, 7]
[1, 2, 3, 4, 5, 6, 7, 8]
[6, 7, 8]

实际输出：

[1]
[1]
[1, 2]
[1, 2]
[1, 2, 3]
[1, 2, 3]
[1, 2, 3, 4]
[1, 2, 3, 4]
[1, 2, 3, 4, 5]
[1, 2, 3, 4, 5]
[1, 2, 3, 4, 5, 6]
[1, 2, 3, 4, 5, 6] // duplicate window even before 30 seconds
[6] // new window after 30 seconds and 1 window from earlier will be dropped
[1, 2, 3, 4, 5, 6, 7]
[6, 7]

因为我在 1 分钟的窗口中创建了 30 秒的希望窗口。我相信，最初应该只有一个窗口，30 秒后应该创建另一个窗口。

有人可以告诉我，实际输出是预期行为还是我遗漏了什么？

注意：我每 4 秒获取一次输入数据，预期/实际输出仅用于表示。

来自 Kafka 文档：

跳跃时间窗口与纪元对齐，间隔较小 bound 是包含的，而上限是排除的。 “与 纪元”表示第一个窗口从时间戳零开始。为了 例如，跳跃窗口大小为 5000ms，提前间隔 3000 毫秒的（“跳跃”）具有可预测的窗口边界 [0;5000),[3000;8000),... - 而不是 [1000;6000),[4000;9000),... 甚至 一些“随机”的东西，例如 [1452;6452),[4452;9452),....

Answer 1

由于您的窗口重叠，每个时间戳都会有多个窗口。对于您的特定窗口配置，您始终会获得 2 个窗口（以毫秒为单位）：

[0,60000)   [60000,12000) [12000,18000) ...
     [30000,90000) [90000,15000) ...

您无法更改此行为，但是，您可以在结果上应用filter()（即aggregate(...).filter(...) 以删除您不感兴趣的窗口。

此外，默认情况下，Kafka Streams 使用记录事件时间。有一个WallclockTimestampExtractor，但它仅在您明确设置时使用。参照。 https://docs.confluent.io/current/streams/developer-guide/config-streams.html#default-timestamp-extractor