使用kafka库反序列化原始AVRO密钥
我目前无法在KSTREAM应用程序中反序列化avro原始密钥
使用avro模式编码的密钥（在模式注册表中注册），
当我使用kafka-avro控制台消费者时，我可以看到密钥是正确的反序列化
但不可能让它在KSTREAM应用程序中工作
密钥的avro模式是一个基元：
{"type":"string"}
我已经遵循了confluent的文档
final Serde<V> valueSpecificAvroSerde = new SpecificAvroSerde<>();
final Map<String, String> serdeConfig = Collections.singletonMap(SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
valueSpecificAvroSerde.configure(serdeConfig, false);
final Serdes.StringSerde keySpecificAvroSerde = new Serdes.StringSerde();
keySpecificAvroSerde.configure(serdeConfig, true);
Consumed<String, totoAvro> inputConf = Consumed.with(keySpecificAvroSerde, valueSpecificAvroSerde);
final KStream<String, totoAvro> mystream = builder.stream("name topic", inputConf);
mystream.peek((key, value) -> logger.info("topic KEY :" + key))
它对值运行良好，但键将是一个字符串，其中包含来自模式注册表的字节，而不仅仅是“卷轴”键
https://docs.confluent.io/current/schema-registry/serializer-formatter.html#wire-format
所以字符串键 /§/./ 11016015201，但我想要卷轴值：1016015201
如果我打印字符串内的字节，则为[0x00 0x00 0x00 0x02 0x31 0x14 0x31 0x30 0x31 0x36 0x30 0x31 0x35 0x32 0x30 0x31]
共有1个答案
使现代化
它现在正在工作：https://stackoverflow.com/a/51957801/6227500
原始答案
该功能当前在架构注册表项目中不可用。
但通过实现自定义SERDE，您可以管理案例，
Thiyaga Rajan提出了一个工作实现
AVRO基元类型的Serde类
-
问题内容： 我目前无法在KSTREAM APP 中 反序列化avro PRIMITIVE密钥 用avro模式编码的密钥（已在模式注册表中注册）， 当我使用kafka-avro-console-consumer时，我可以看到密钥已正确反序列化 但是不可能使其在KSTREAM应用程序中工作 密钥的avro模式是主要的： 我已经关注了合流的文档 它对于该值工作得很好，但是该键将是一个字符串，该字符串包含
-
null null
-
主要目标是聚合两个Kafka主题，一个压缩慢速移动数据，另一个每秒接收一次的快速移动数据。 我已经能够在简单的场景中使用消息，例如KV（Long， String），使用如下内容： 但是，当您需要从 AVRO 反序列化时，这似乎不是方法。我有一个KV（字符串，AVRO），我需要消费。 我尝试从AVRO模式生成Java类，然后将它们包含在“应用”中，例如: 但这似乎不是正确的方法。 是否有任何文档/示
-
我正在用Kafka、星火和朱皮特笔记本做概念验证，我遇到了一个奇怪的问题。我正在试着阅读从Kafka到Pyspark的Avro记录。我正在使用汇合模式注册表获取模式以反序列化avro消息。反序列化spark dataframe中的avro消息后，结果列为空，没有任何错误。列应该包含数据，因为当强制转换为字符串时，某些avro字段是可读的。 我也尝试过在Scala中的spark-shell（没有ju
-
我是Avro和Kafka的新手，我花了几天时间来发送关于Kafka主题的序列化数据...不成功。 让我来解释一下我想要达到的目标： 在生产者方面，我通过SOAP接收数据并发送关于Kafka主题的内容。我正在使用CXF从WSDL生成POJO，并且编写了相应的模式。我正在尝试做的是序列化由CXF解封的对象，并在我的Kafka主题上发送它们。 在web上找到的大多数示例中，Avro记录都是使用已知的模式