Как сгенерировать pojo из записи kafka avro?

У меня есть класс User, и я разделяю его как avro (используя сериализатор Confluent avro и реестр схем) и публикую его в теме Kafka. Я заставил потребителя печатать данные на консоль, и он отлично работает. Сейчас я пытаюсь создать исходный объект из этих данных. Например, я публикую объект «Пользователь» как avro в теме Kafka. Я пытаюсь воссоздать этот пользовательский объект (вместо вывода на консоль) после его использования. Это возможно?

Ниже мой код

Класс пользователя

public class User { 
    int id;
    String name;    
    public User(){} 
    public User(int id, String name) {
        super();
        this.id = id;
        this.name = name;
    }

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }   
}

Потребительский код

User user = new User();



Properties props = new Properties();

props.put("bootstrap.servers", "127.0.0.1:9092");

props.put("group.id", "avro-consumer-group");

props.put("key.deserializer", io.confluent.kafka.serializers.KafkaAvroDeserializer.class);

props.put("value.deserializer", io.confluent.kafka.serializers.KafkaAvroDeserializer.class);

props.put("schema.registry.url","http://127.0.0.1:8081");

KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<String, GenericRecord>(props);



consumer.subscribe(Arrays.asList("avrotesttopic"));

System.out.println("Subscribed to topic " + "avrotesttopic");



while (true) {

    ConsumerRecords<String, GenericRecord> records = consumer.poll(100);

    for (org.apache.kafka.clients.consumer.ConsumerRecord<String, GenericRecord> record : records){

        System.out.printf("value = %sn",record.value());    

        //output->  value = {"id": 10, "name": "testName"}

    }

}

Спасибо


java apache-kafka avro confluent-schema-registry
person Alfred    schedule 04.05.2018    source источник

Ответы (3)


arrow_upward
4
arrow_downward

Учитывая, что вы используете KafkaAvroDeserializer, вам нужно будет установить следующее свойство как часть конфигурации вашего потребителя.

    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");

это позволит вам получить SpecificRecord вместо GenericRecord, уже обработанных Потребителем. Вот пример

https://dzone.com/articles/kafka-avro-serialization-and-the-schema-registry

person hlagos    schedule 04.05.2018

arrow_upward
0
arrow_downward

Вы можете преобразовать AvroSchema в JsonSchema, а затем из JsonSchema вы можете создать класс POJO.

Конвертировать AVRO в JSON:

static void convertAvroToJson(InputStream inputStream, OutputStream outputStream, Schema schema)
        throws IOException {
    DatumReader<Object> reader = new GenericDatumReader<>(schema);
    DatumWriter<Object> writer = new GenericDatumWriter<>(schema);

    BinaryDecoder binaryDecoder = DecoderFactory.get().binaryDecoder(inputStream, null);

    JsonEncoder jsonEncoder = EncoderFactory.get().jsonEncoder(schema, outputStream, true);
    Object datum = null;
    while (!binaryDecoder.isEnd()) {
        datum = reader.read(datum, binaryDecoder);
        writer.write(datum, jsonEncoder);
        jsonEncoder.flush();
    }
    outputStream.flush();
}

А затем преобразовать JSONSchema в POJO,

https://dzone.com/articles/converting-json-to-pojos-using-java

person ideano1    schedule 04.05.2018

arrow_upward
0
arrow_downward

Это зависит от того, как вы сериализуете входные записи в Avro.

Короткий ответ, если ваша сериализация в порядке, вы должны внести изменения ниже при получении записей

ConsumerRecords<String, User> records = consumer.poll(100);

Полный пример см. В этом

person asolanki    schedule 04.05.2018