Kafka Streams：reduce操作中如何获取key？或实现指定聚合输出逻辑

首先明确：reduce方法本身无法直接获取Key，它的参数只有聚合值和当前输入值，没有Key相关参数。不过可以通过以下几种方式实现你的需求：

方案一：使用aggregate替代reduce #

aggregate的聚合逻辑（Aggregator）可以直接获取Key，是最适合的方案。我们可以维护一个布尔状态，标记当前Key对应的所有Value是否都等于Key，最后过滤出状态为true的结果输出。

示例代码（Java）：

KStream<String, String> inputStream = ...;

inputStream
    .groupByKey()
    .aggregate(
        // 初始化状态：默认认为符合条件（还未遇到不符合的Value）
        () -> true,
        // 聚合逻辑：每次新Value进来，判断是否等于Key，同时保留之前的状态
        (key, value, isValid) -> isValid && value.equals(key),
        // 指定状态存储的名称
        Materialized.as("valid-key-status-store")
    )
    // 只保留所有Value都等于Key的Key
    .filter((key, isValid) -> isValid)
    .toStream()
    // 输出结果，这里可以根据需求选择输出Key或其他格式
    .to("qualified-output-topic");

方案二：将Key嵌入Value中再做reduce #

如果一定要用reduce，可以先把Key和原Value封装成一个对象，这样在reduce时就能从Value中取出Key进行判断。需要注意这个封装对象需要支持Kafka的序列化/反序列化。

示例代码（Java）：

// 先定义封装Key和Value的类（需实现Serde）
class KeyWrappedValue {
    private String key;
    private String value;

    // 构造器、getter、序列化相关方法省略
}

KStream<String, String> inputStream = ...;

inputStream
    // 将Key嵌入Value对象中
    .map((key, value) -> KeyValue.pair(key, new KeyWrappedValue(key, value)))
    .groupByKey()
    .reduce(
        (agg, curr) -> {
            // 判断当前Value是否等于Key，且之前的聚合结果也符合条件
            boolean stillValid = agg.getValue().equals(agg.getKey()) && curr.getValue().equals(curr.getKey());
            // 返回新的聚合对象，若不符合则标记为无效
            return stillValid ? curr : new KeyWrappedValue(curr.getKey(), null);
        },
        Materialized.as("key-wrapped-store")
    )
    // 过滤出符合条件的结果
    .filter((key, wrapped) -> wrapped.getValue() != null && wrapped.getValue().equals(key))
    .toStream()
    .map((key, wrapped) -> KeyValue.pair(key, key))
    .to("qualified-output-topic");

方案三：使用process/transformValues自定义状态管理 #

如果需要更灵活的控制，可以使用process API自定义处理器，直接在处理逻辑中获取Key，并手动维护状态记录是否符合条件。这种方式代码量较大，但适合复杂场景。

内容的提问来源于stack exchange，提问作者simonalexander2005