我们使用的是带有模式注册表的合流Kafka。我们有40多个主题。我们的应用程序使用注册表中的模式将avro消息写入这些主题。
据我所知,在使用注册表时,消息并不包含实际的模式,而是对注册表中的模式(schema id)的引用。我正在用java开发一个实用程序,它将接受一个主题名或一个主题名列表,并从每个主题中检索有限数量(可能是50条)的消息。然后,该实用程序应将avro消息转换为json。
我看到的所有示例都必须给出能够将byte[]转换为json的模式。
我希望并且想知道,是否有可能使用来自主题的avro消息(schema id)的信息从注册表中动态地获取模式,并使用该信息将消息转换为json。
有可能吗?有人能给我举个例子说明如何做到这一点吗?
谢谢您
更新
@艾克
试验1
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer");
final Consumer<String, GenericRecord> genericConsumer = new KafkaConsumer<>(props);
genericConsumer.subscribe(Collections.singletonList("TOPICNAME"));
while (true) {
final ConsumerRecords<String, GenericRecord> genericConsumerRecords = genericConsumer.poll(Duration.ofMillis(1000));
System.out.println("genericConsumerRecords.count() : " + genericConsumerRecords.count() + " genericConsumerRecords.isEmpty() : " + genericConsumerRecords.isEmpty());
genericConsumerRecords.forEach(genericRecord1 -> {
try {
System.out.println("convert(genericRecord1.value()) -> " + convert(genericRecord1.value()));
} catch (IOException e) {
e.printStackTrace();
}
//
});
}
这是输出 genericConsumerRecords.count() : 0 genericConsumerRecords.isEmpty() : true
注意:convert方法是下面答案中的方法
试验2
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.ByteArrayDeserializer");
final Consumer<String, byte[]> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("TOPICNAME"));
while (true) {
final ConsumerRecords<String, byte[]> consumerRecords = consumer.poll(Duration.ofMillis(1000));
System.out.println("consumerRecords.count() : " + consumerRecords.count() + " consumerRecords.isEmpty() : " + consumerRecords.isEmpty());
consumerRecords.forEach(record1 -> {
String string = new String(record1.value(), StandardCharsets.UTF_8);
System.out.printf("offset = %d, key = %s, value = %s \n", record1.offset(), record1.key(), string);
});
}
这是输出
consumerRecords.count() : 60 consumerRecords.isEmpty() : false
offset = 0, key = e3bff195-08a7-4c58-99de-98ffe2d460e6, value = He52d6fa6-841f-430c-8bf7-bd4c7b684129 http://schemaregistryurl:8081/subjects/TOPICNAME-value/versions/1/schema Canon Message to represent CustomerPrefAVRFAST 162019-08-07T08:35:35.9950728 QA1-Test-0421-16$CustomerPrefData 1He52d6fa6-841f-430c-8bf7-bd4c7b684129 RawH862437d0-e260-45f9-ab5e-345b536d685a02020-04-21T17:48:52.601Z$CustomerPref POL_MAST02020-04-21T11:17:28.241ZHe3bff195-08a7-4c58-99de-98ffe2d460e69
False&1900-01-01T00:00:00He3bff195-08a7-4c58-99de-98ffe2d460e6He3bff195-08a7-4c58-99de-98ffe2d460e6
注意:我不得不从输出中删除一些非ascii字符
第二个方法确实给出了输出,但它是一个字节[],需要获得json输出。尝试过不同的方法都没有成功。
我该怎么修?
谢谢
1条答案
按热度按时间z5btuh9x1#
一种选择是使用
io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer
获取标准的apache avroGenericRecord
然后做:以json格式返回消息。架构由检索
record.getSchema()
.