Kafka Streams 取输入主题4个Double值计算后按键聚合的语法问题

报错原因与解决方案 #

1. aggregate方法类型推断失败 #

该报错是由于初始值类型、聚合器返回值类型、Materialized指定的序列化类型三者不统一导致：你之前的初始值{0L}为Long类型，但自定义函数返回Double，同时替换了值序列化器为Double类型，类型不匹配触发推断失败。
修正方法：统一三者类型为Double即可，示例如下：

.aggregate(
    { 0.0 }, // 初始值改为Double类型的0.0
    { _, trip, total -> total + exampleFunction(trip) }, // 聚合器输入输出均为Double
    Materialized.with(Serdes.String(), Serdes.Double()) // 值序列化器使用Double类型
)

如果需要从Trip对象提取4个Double值传入自定义函数，直接在聚合器中读取对应字段调用即可：

{ _, trip, total -> 
    // 提取4个Double字段传入自定义函数
    val funcRes = yourCustomFunction(trip.field1, trip.field2, trip.field3, trip.field4)
    total + funcRes
}

2. toStream写入输出主题报错 #

该报错的核心原因是你使用了会话窗口聚合（代码中指定了SessionStore作为状态存储，同时传入了merger合并器参数），窗口聚合后的KTable的key类型不是普通String，而是封装了原始key和窗口时间范围的Windowed<String>，所以toStream后的KStream key类型为Windowed<String>，和你传入的Serdes.String()类型不匹配。

解决方法1：保留窗口信息写入

直接使用窗口类型的序列化器即可：

import org.apache.kafka.streams.kstream.WindowedSerdes

// 写入时指定key为会话窗口String类型的Serde
distance.toStream().to(
    outputTopic, 
    Produced.with(
        WindowedSerdes.sessionWindowedSerdeFrom(String::class.java),
        Serdes.Double()
    )
)

解决方法2：提取原始key写入（不需要保留窗口信息）

如果不需要存储窗口时间范围，先将Windowed<String>转换为原始的String key后再写入：

distance.toStream()
    // 提取Windowed对象中的原始业务key
    .map { windowedKey, value -> KeyValue.pair(windowedKey.key(), value) }
    .to(outputTopic, Produced.with(Serdes.String(), Serdes.Double()))

内容的提问来源于stack exchange，提问作者Phil O'Dowd