时间戳

首页 > 解决方案 > 聚合上使用了错误的序列化程序

apache-kafka-streams - 聚合上使用了错误的序列化程序

问题描述

我正在处理一个处理日志事件的 kafka-streams 应用程序。在这种情况下，我想将 WorkflowInput 类型聚合为 Workflow 类型。我在使聚合工作时遇到问题。

final KStream<String, WorkflowInput> filteredStream = someStream;
final KTable<String, Workflow> aggregatedWorkflows = filteredStream
    .peek((k, v) -> {
        if (!(v instanceof WorkflowInput)) {
            throw new AssertionError("Type not expected");
        }
    })
    .groupByKey()
    .<Workflow>aggregate(Workflow::new, (k, input, workflow) -> workflow.updateFrom(input),
            Materialized.<String, Workflow, KeyValueStore<Bytes, byte[]>>as("worflow-cache")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.serdeFrom(new JsonSerializer<Workflow>(), new JsonDeserializer<Workflow>(Workflow.class)));

我得到以下异常：由以下原因引起：org.apache.kafka.streams.errors.StreamsException: A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.StringSerializer) is not compatible to the actual key or value type (key type: java.lang.String / value type: workflowauditstreamer.WorkflowInput).

有两件事需要注意： * 值序列化器是一个 StringSerializer，而我使用withValueSerde. * 实际值类型是WorkflowInput我所期望的Workflow，因为那是我的聚合值类型。

我是 kafka-streams 的新手，所以我可能会遗漏一些明显的东西，但我无法弄清楚。我在这里想念什么？

标签： apache-kafka-streams

解决方案

如果您Serde从配置中覆盖默认值，则它在操作员就地覆盖。它不会向下游传播（Kafka 2.0——有 WIP 来改进它）。

因此，您也需要将Serde您使用的 s传递someStream = builder.stream(...)到.groupByKey(Serialized.with(...))中。

推荐阅读