Kafka流由聚合窗口化，并在记忆中保留

Question

我有一个KStream

java.lang.OutOfMemoryError: Java heap space

KStream DSL如下所示：

TimeWindows timeWindows = TimeWindows.of(Duration.ofDays(1)).advanceBy(Duration.ofMillis(1));
Initializer<History> historyInitializer = History::new;
        Aggregator<String, Event, History> historyAggregator = (key, value, aggregate) -> {
            aggregate.key = value.uuid;
            aggregate.addHistoryEventWindow(value);
            return aggregate;
        };

KTable<String, History> historyWindowed = eventStreamRaw
.filter((key, value) -> value != null)
    .groupByKey(Grouped.with(Serdes.String(), this.eventSerde))
    // segment our messages into 1-day windows
    .windowedBy(timeWindows)
    .aggregate(historyInitializer, historyAggregator, Named.as("name"), Materialized.with(Serdes.String(), this.historySerde))
    .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
    .groupBy(
            (key, value) -> new KeyValue<String, History>(
                    value.key + "|+|" + key.window().start() + "|+|" + key.window().end(), value),
            Grouped.with(Serdes.String(), this.historySerde))
    .aggregate(History::new, (key, value, aggValue) -> value, (key, value, aggValue) -> value,
            Materialized.with(Serdes.String(), this.historySerde));

阅读一些文章（例如Kafka流窗口）

但我想补充一点，这对我来说并不适用：

final Materialized<String, History, WindowStore<Bytes, byte[]>> store = Materialized.<String, History, WindowStore<Bytes, byte[]>>as("eventstore")
        .withKeySerde(Serdes.String())
        .withValueSerde(this.historySerde)
        .withRetention(Duration.ofDays(1).plus(Duration.ofMillis(1)));

KTable<String, History> historyWindowed = eventStreamRaw
    ...
    .aggregate(historyInitializer, historyAggregator, Named.as("name"), store)

Java编译器抛出以下错误：

The method 
aggregate(Initializer<VR>, Aggregator<? super String,? super Event,VR>, Named, Materialized<String,VR,WindowStore<Bytes,byte[]>>) 
in the type TimeWindowedKStream<String,Event> is not applicable for the arguments 
(Initializer<History>, Aggregator<String,Event,History>, Named, Materialized<String,History,WindowStore<Bytes,byte[]>>)

老实说，我不明白。参数是正确的；虚拟现实类型是“历史”。

你知道我错过了什么吗？

这个windowedBy KTable的想法是让一个状态为一件“事情”保存所有事件一天。假设生成了一个新警报，我想将一天内“某物”的所有事件附加到警报上。然后，我将从KStream警报到KTable历史进行leftJoin。这是为Kafka事件添加历史数据的最佳方式吗？有没有办法“查找”KStream活动的最后x天？我已经检查了KStream Alert KStream事件leftJoin，但这将为每个新的KStream事件生成一个输出。因此，从我的观点来看，这是不可行的。

非常感谢你的帮助。我希望这只是一个简单的修复。非常感谢！

Answer 1

看看下面的后Kafka流应用程序-计数和总和聚合，我导入了错误的“字节”类。因此，请确保导入以下类“org.apache.kafka.common.utils.Bytes”。

但是，也许你有一个更好的主意，用一个外键关联的另一个流的历史数据来丰富来自一个流的Kafka消息。

谢谢伙计们。