Как извлечь схему AVRO из реестра схем из сообщения в Topi c от java потребителя - PullRequest
Новая
       80

Как извлечь схему AVRO из реестра схем из сообщения в Topi c от java потребителя

0 голосов
/ 26 апреля 2020

Мы используем Confluent Kafka с реестром схемы. У нас более 40 тем. Наше приложение пишет сообщения avro в эти темы, используя схему из реестра.

Из того, что я понимаю, при использовании реестра сообщение не содержит фактическую схему, а содержит ссылку на схему (идентификатор схемы) из реестра. Я работаю над утилитой в Java, которая примет имя topi c или список имен topi c и получит ограниченное количество (возможно, 50) сообщений из каждой из этих тем. Эта утилита должна затем преобразовать сообщение avro в json.

Все примеры, которые я видел, все еще должны дать схему для возможности преобразования байта [] в json.

Я надеялся и задавался вопросом, возможно ли это чтобы получить схему из реестра динамически, используя информацию из сообщения avro (идентификатор схемы) из topi c и использовать ее для преобразования сообщения в json.

Возможно ли это сделать ? Может кто-нибудь привести пример того, как этого добиться?

Спасибо

Обновления

@ eik

Пробная версия 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

Примечание: : метод преобразования - тот, который в ответ ниже

Trial 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 Ответ

0 голосов
/ 28 апреля 2020

Одним из вариантов будет использование io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer для получения стандартного Apache Avro GenericRecord и затем: 

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;

class GenericRecord2Json {
  String convert(GenericRecord record) throws IOException {
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

    Schema schema = record.getSchema();
    DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(schema);
    Encoder encoder = EncoderFactory.get().jsonEncoder(schema, outputStream, /* pretty= */ true);
    writer.write(record, encoder);
    encoder.flush();

    return outputStream.toString(StandardCharsets.UTF_8);
  }
}

, который возвращает ваше сообщение в формате JSON. Схема извлекается record.getSchema().

...