Как исправить ожидаемый start-union. Получил VALUE_NUMBER_INT при преобразовании JSON в Avro в командной строке?

Ответ 1

В соответствии с объяснением Дуга Рединг,

Для кодирования Avro JSON требуется, чтобы были отмечены ненулевые значения объединения с их предполагаемым типом. Это потому, что профсоюзы [ "bytes", "string" ] и [ "int", "long" ] неоднозначны в JSON, первом оба кодируются как строки JSON, а вторая кодируются как JSON.

http://avro.apache.org/docs/current/spec.html#json_encoding

Таким образом, ваша запись должна быть закодирована как:

{"name": "Alyssa", "favorite_number": {"int": 7}, "favorite_color": null}

Ответ 2

В работе есть новый кодер JSON, который должен решить эту общую проблему:

https://issues.apache.org/jira/browse/AVRO-1582

https://github.com/zolyfarkas/avro

Ответ 3

Как отметил @Emre-Sevinc, проблема заключается в кодировке вашей записи Avro.

Чтобы быть более конкретным здесь;

Не делай этого:

   jsonRecord = avroGenericRecord.toString

Вместо этого сделайте это:

    val writer = new GenericDatumWriter[GenericRecord](avroSchema)
    val baos = new ByteArrayOutputStream
    val jsonEncoder = EncoderFactory.get.jsonEncoder(avroSchema, baos)
    writer.write(avroGenericRecord, jsonEncoder)
    jsonEncoder.flush

    val jsonRecord = baos.toString("UTF-8")

Вам также понадобится следующий импорт:

import org.apache.avro.Schema
import org.apache.avro.generic.{GenericData, GenericDatumReader, GenericDatumWriter, GenericRecord}
import org.apache.avro.io.{DecoderFactory, EncoderFactory}

После этого вы получите jsonRecord с ненулевыми значениями объединения, помеченными их предполагаемым типом.

Надеюсь это поможет !

Ответ 4

Я реализовал объединение и его проверку, просто создал схему объединения и передал ее значения почтальону. Resgistry url - это URL, который вы указываете для свойств kafka, вы также можете передавать динамические значения в вашу схему

RestTemplate template = new RestTemplate();
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        HttpEntity<String> entity = new HttpEntity<String>(headers);
        ResponseEntity<String> response = template.exchange(""+registryUrl+"/subjects/"+topic+"/versions/"+version+"", HttpMethod.GET, entity, String.class);
        String responseData = response.getBody();
        JSONObject jsonObject = new JSONObject(responseData);
        JSONObject jsonObjectResult = new JSONObject(jsonResult);
        String getData = jsonObject.get("schema").toString();
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(getData);
        GenericRecord genericRecord = new GenericData.Record(schema);
        schema.getFields().stream().forEach(field->{
            genericRecord.put(field.name(),jsonObjectResult.get(field.name()));
        });
        GenericDatumReader<GenericRecord>reader = new GenericDatumReader<GenericRecord>(schema);
        boolean data = reader.getData().validate(schema,genericRecord );