如何在流聚合中指定Serdes？

Question

我正在开发一个Kafka streams应用程序，我遇到了一些麻烦，想知道如何使聚合工作。

我有一个KStreamBankTransactions，其中键的类型为String，值的类型为JsonNode，因此我使用

// Definition of the different Serdes used in the streams
final Serde<String> stringSerde = Serdes.String();
final Serde<JsonNode> jsonSerde = new JsonSerde();
final Serde<Long> longSerde = Serdes.Long();

config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, stringSerde.getClass().getName());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, jsonSerde.getClass().getName());

我希望将值聚合在ktable 中，其中键将相同，但值将是从我的JSON中提取的long。

所以首先我写到：

KTable<String, Long> totalBalances = bankTransactions
        .groupByKey()
        .aggregate(
                () -> 0L,
                (key, transaction, balance) -> (Long)((Long)balance + transaction.get("amount").asLong()),
                Materialized.as("bank-total-balance")
        );

并且在运行时会出现以下错误：

Caused by: org.apache.kafka.streams.errors.StreamsException:
A serializer (value: org.apache.kafka.connect.json.JsonSerializer) is not compatible to
the actual value type (value type: java.lang.Long).
Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.

我知道Kafka在抱怨，因为我试图使用缺省的Json serdes序列化long。所以从汇流公司的医生那里我试了一下

KTable<String, Long> totalBalances = bankTransactions
        .groupByKey()
        .aggregate(
                () -> 0L,
                (key, transaction, balance) -> (Long)((Long)balance + transaction.get("amount").asLong()),
                Materialized.as("bank-total-balance").withValueSerde(Serdes.Long())
        );

但我在编译时遇到了一个错误：

Error:(121, 89) java: incompatible types:
org.apache.kafka.common.serialization.Serde<java.lang.Long> cannot be converted
to org.apache.kafka.common.serialization.Serde<java.lang.Object>

我尝试了不同的方法来编写这段代码（例如，使用serdes.long()而不是我的longserdes，尝试参数化materialize的类型，甚至尝试将我的初始化器和聚合器写成函数（Java7风格），但我不知道我做错了什么。

因此，我的问题很简单：当aggregate不是默认SERDE时，如何正确指定aggregate应该使用的SERDE？

Answer 1

正确的语法似乎如下：

KTable<String, Long> totalBalances = bankTransactions
        .groupByKey()
        .aggregate(
                () -> 0L,
                (key, transaction, balance) -> (Long)((Long)balance + transaction.get("amount").asLong()),
                Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("bank-total-balances")
                        .withKeySerde(stringSerde)
                        .withValueSerde(longSerde)
        );

materialize.之后的三种类型是key类型、value类型和用于物化KTable的store类型，这个类型不应该改变。然后我们可以在这个键值存储中定义用于写入的Serdes。

注我从github上的一个随机回购中得到了这个语法，我仍然很乐意接受一个有一些文档支持的更精确的答案。