使用Kafka流将状态持久化到Kafka

Question

我试着把我的头缠绕在Kafka的溪流和一些根本的问题，我似乎无法解决，我自己。我理解ktable和Kafka状态存储的概念，但我很难决定如何实现它。我还在使用Spring Cloud Streams，这在此基础上增加了另一个层次的复杂性。

我的用例：

@Bean
public Function<KStream<String, ProcessNode>, KStream<String, List<IndicatorEvaluation>>> process() {
    return input -> input.mapValues(this::analyze).filter((host, evaluation) -> evaluation != null);
}

public List<IndicatorEvaluation> analyze(final String host, final ProcessNode process) {
    // Does stuff
}

一些有状态规则如下所示：

[some condition] REPEATS 5 TIMES WITHIN 1 MINUTE
[some condition] FOLLOWEDBY [some condition] WITHIN 1 MINUTE
[rule A exists and rule B exists]

我当前的实现将所有这些信息存储在内存中，以便能够执行分析。由于显而易见的原因，它不容易扩展。所以我想我会坚持到Kafka的州立商店。

我不确定做这件事的最好办法。我知道有一种方法可以创建自定义状态存储，从而允许更高级别的灵活性。我不确定KafkaDSL是否支持这一点。

Answer 1

从您给出的描述来看，我相信这个用例仍然可以在Kafka流中使用DSL来实现。上面显示的代码不跟踪任何状态。在您的拓扑中，您需要通过跟踪规则的计数来添加状态，并将它们存储在状态存储区中。则只需要在计数达到阈值时发送输出规则。下面是这个作为伪代码背后的大致思路。显然，您必须对其进行调整，以满足您的用例的特定规范。

@Bean
public Function<KStream<String, ProcessNode>, KStream<String, List<IndicatorEvaluation>>> process() {
    return input -> input
                     .mapValues(this::analyze)
                     .filter((host, evaluation) -> evaluation != null)
                     ...
                     .groupByKey(...)
                     .windowedBy(TimeWindows.of(Duration.ofHours(1)))
                     .count(Materialized.as("rules"))
                     .filter((key, value) -> value > 4)
                     .toStream()
                    ....
}