序列化程序与org.apache.avro.util.Utf8不兼容

Question

我试图使用kafkastreams进行聚合，但得到的错误如下所示

这是我正在做的事情：

KGroupedStream<String, Long> countrywiseAmount = ......;
KTable<String, CountSum> countrywiseAverageSum = countrywiseAmount
                .aggregate(new Initializer<CountSum>() {
                            @Override
                            public CountSum apply() {
                                return new CountSum();
                            }
                }, new Aggregator<String,Long,CountSum>() {
                    @Override
                    public CountSum apply(String country, Long amount, CountSum sumByCountry) {
                        sumByCountry.setCountry(country.toString());
                        sumByCountry.setCount(sumByCountry.getCount()+1);
                        sumByCountry.setSum(sumByCountry.getSum()+amount);
                        return sumByCountry;
                    }
                }, Materialized.with(stringSerde, countSumSerde));

我收到的错误如下。

由以下原因引起：A 序列化程序（密钥：类型：a 序列化程序（密钥：类型：网站名称：通用序列化程序 / 值：在流配置中更改默认的 Serdes 或通过方法参数提供正确的 Serdes。在 org.apache.kafka.流.处理器.内部.sinkNode.进程 （SinkNode.java：94） 在 org.apache.kafka.流.处理器.internals.处理器上下文前进（处理器上下文.java：201） 在 org.apache.kafka.streams.processor.internals.processorContextul.forward（处理器上下文.java：180） 在 org.apache.kafka.流媒体.处理器.内部.处理器上下文简单处理器（简单.java：133） 在在 org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process（KStreamFilter.java：43） at org.apache.kafka.流媒体.processor.internals.processorNode.process（ProcessorNode.java：117） at org.apache.kafka.streams.processor.processor.internals.internals..java 处理器.java：180） at org.apache.kafka.流.处理器.内部.处理器上下文.java：180） at在组织.apache.kafka.streams.processor.internals.processorContextins.forward（处理器上下文过程.java：133） at org.apache.kafka.streams.kstream.internals.kStreamMapor.process（KStreamMap.java：42） at org.apache.kafka.streams.processor.internals.processorNode.process（ProcessorNode.java：117） at org.apache.kafka.stream.processor.internals.ProcessorContextimpl.forward（ProcessorContextimpl.java：201）处理器网明.java：133） 在 org.java.apache.kaf.java ka.streams.processor.internals.process（流任务.java：363） ...5 更多 原因： java.lang.ClassCastException： org.apache.avro.util.Utf8 不能被转换为 java.lang.String 在 org.apache.kafache.common.serialization.字符串序列化（字符串序列化程序.java：28） 在 org.apache.kafka.common.serialization.serializer.serialize（序列化程序.java：62） 在 org.apache.kafache.流.处理器.internals.内部体.记录收集器发送（记录收集器.java：162） 在 org.apache.kafka.流.内部.记录收集器发送（记录收集器.java：103） at org.apache.kafka.stream.internals.sinkNode.process（SinkNode.java：89）

有线索吗？

Answer 1

KTable聚合逻辑没有问题，但是KGroupeStream的问题