主页 / coding / 问题 / 79562196
Accepted
simonalexander2005
Asked: 2025-04-08 21:35:55 +0800 CST

KGroupedStream reduce 在 Reducer 逻辑中使用 key

  • 772

我正在使用 Kafka Streams 对 kafka 主题进行分组和减少。

如果针对给定键的所有值，键和值都相等，我想为该键生成一个输出；否则，不要为该键输出任何内容。

然而，要做到这一点，我需要访问key，并且.reduce方法只允许访问聚合/当前值。

调用 .reduce 时可以获取密钥吗？或者还有其他方法吗？

apache-kafka

1 个回答

  1. Best Answer

    您可以使用接口aggregate的方法KGroupedStream：

    <VR> KTable<K,VR> aggregate(Initializer<VR> initializer, Aggregator<? super K,? super V,VR> aggregator)

    按分组的键聚合此流中记录的值。包含null键或值的记录将被忽略。聚合是泛化操作combining via reduce(...)，例如，它允许结果具有与输入值不同的类型。

    如果您希望仅当值匹配时才输出键的结果，则可以创建一个 Ktable，如下所示：

    KTable<String, Boolean> resultTable = inputStream
    .groupByKey()
    .aggregate(
        () -> true,
        (key, value, agg) -> agg && key.equals(value),
        Materialized.with(Serdes.String(), Serdes.Boolean())
    );

    然后您可以过滤表中的值，例如：

    KStream<String, Boolean> finalResult = resultTable
    .toStream()
    .filter((key, allEqual) -> allEqual);
    • 1

相关问题