Avro可选枚举字段转GenericRecord报错问题及无依赖解决方案咨询

原本可以通过给定的Avro Schema和JSON数据生成GenericRecord，但将Schema中的gender字段修改为包含null的联合类型（可选枚举字段）后，程序抛出AvroTypeException错误。需要在不使用allegro/json-avro-converter第三方工具的前提下解决该问题，并提供代码示例。

原Avro Schema #

{
  "type": "record",
  "name": "Person",
  "fields": [
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "age",
      "type": "int"
    },
    {
      "name": "city",
      "type": "string"
    },
    {
      "name": "gender",
      "type": {
        "type": "enum",
        "name": "Gender",
        "symbols": ["MALE", "FEMALE"]
      }
    }
  ]
}

原JSON数据 #

{"name": "John", "age": 30, "city": "New York", "gender": "MALE"}

原Java代码 #

public class Main {

    public static void main(String[] args) throws IOException {
        Schema schema = readSchema();
        JsonNode data = readData();
        GenericRecord genericRecord = convertJsonToAvro(data, schema);
        System.out.println(genericRecord);
    }

    public static GenericRecord convertJsonToAvro(JsonNode jsonNode, Schema avroSchema) throws IOException {
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(avroSchema);
        Decoder decoder = DecoderFactory.get().jsonDecoder(avroSchema, jsonNode.toString());
        return reader.read(null, decoder);
    }

    private static Schema readSchema() throws IOException {
        InputStream inputStream = Main.class.getClassLoader().getResourceAsStream("schemas/person.avsc");
        return new Schema.Parser().parse(inputStream);

    }

    private static JsonNode readData() throws IOException {
        InputStream inputStream = Main.class.getClassLoader().getResourceAsStream("sample_data/person.json");
        ObjectMapper objectMapper = new ObjectMapper();
        return objectMapper.readValue(inputStream, JsonNode.class);
    }

}

修改后的可选gender字段定义 #

{
    "name": "gender",
    "type": [
        "null",
        {
            "type": "enum",
            "name": "Gender",
            "symbols": ["MALE", "FEMALE"]
        }
    ],
    "default": null
}

报错信息 #

Exception in thread "main" org.apache.avro.AvroTypeException: Expected start-union. Got VALUE_STRING
at org.apache.avro.io.JsonDecoder.error(JsonDecoder.java:511)
at org.apache.avro.io.JsonDecoder.readIndex(JsonDecoder.java:430)
at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:282)
at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:188)
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:161)
at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:260)
at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:248)
at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:180)
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:161)
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:154)
at com.generic.Main.convertJsonToAvro(Main.java:27)
at com.generic.Main.main(Main.java:20)

FAILURE: Build failed with an exception.

使用的依赖 #

implementation group: 'org.apache.avro', name: 'avro', version: '1.11.3'
implementation group: 'com.fasterxml.jackson.core', name: 'jackson-core', version: '2.16.0'

报错原因是Avro的JSON解码器对联合类型有严格格式要求：当字段是[null, 枚举]的联合类型时，JSON数据中的该字段需要用带类型标记的对象格式，而非直接传递枚举字符串。比如原JSON中的"gender": "MALE"需改为"gender": {"Gender": "MALE"}；若为null值，直接写"gender": null即可。

如果不想修改原始JSON文件，可通过代码预处理JSON节点，将联合类型字段转换为Avro要求的格式，以下是修改后的完整代码：

修改后的Java代码 #

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.IOException;
import java.io.InputStream;

public class Main {

    public static void main(String[] args) throws IOException {
        Schema schema = readSchema();
        JsonNode data = readData();
        // 预处理JSON数据，适配Avro联合类型格式
        JsonNode processedData = processUnionFields(data, schema);
        GenericRecord genericRecord = convertJsonToAvro(processedData, schema);
        System.out.println(genericRecord);
    }

    public static GenericRecord convertJsonToAvro(JsonNode jsonNode, Schema avroSchema) throws IOException {
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(avroSchema);
        Decoder decoder = DecoderFactory.get().jsonDecoder(avroSchema, jsonNode.toString());
        return reader.read(null, decoder);
    }

    private static Schema readSchema() throws IOException {
        InputStream inputStream = Main.class.getClassLoader().getResourceAsStream("schemas/person.avsc");
        return new Schema.Parser().parse(inputStream);
    }

    private static JsonNode readData() throws IOException {
        InputStream inputStream = Main.class.getClassLoader().getResourceAsStream("sample_data/person.json");
        ObjectMapper objectMapper = new ObjectMapper();
        return objectMapper.readValue(inputStream, JsonNode.class);
    }

    // 预处理JSON节点，将[null, 枚举]类型的字段转换为Avro要求的格式
    private static JsonNode processUnionFields(JsonNode jsonNode, Schema schema) {
        if (schema.getType() != Schema.Type.RECORD) {
            return jsonNode;
        }
        ObjectNode processedNode = ((ObjectNode) jsonNode).deepCopy();
        for (Schema.Field field : schema.getFields()) {
            Schema fieldSchema = field.schema();
            JsonNode fieldValue = processedNode.get(field.name());
            if (fieldValue == null || fieldValue.isNull()) {
                continue;
            }
            // 判断字段是否为包含null和枚举的联合类型
            if (fieldSchema.getType() == Schema.Type.UNION) {
                Schema enumType = null;
                for (Schema unionType : fieldSchema.getTypes()) {
                    if (unionType.getType() != Schema.Type.NULL) {
                        enumType = unionType;
                        break;
                    }
                }
                if (enumType != null && enumType.getType() == Schema.Type.ENUM) {
                    // 将枚举字符串转换为带类型标记的对象
                    ObjectNode enumNode = new ObjectMapper().createObjectNode();
                    enumNode.put(enumType.getName(), fieldValue.asText());
                    processedNode.set(field.name(), enumNode);
                }
            }
        }
        return processedNode;
    }
}

代码说明 #

1. 新增的processUnionFields方法会遍历Record的所有字段，识别出[null, 枚举]类型的联合字段，将原始枚举字符串转换为{"枚举名": "枚举值"}的格式，适配Avro解码器要求。
2. 若JSON中gender字段为null，代码会直接保留，无需转换——Avro支持直接解析null作为联合类型的第一个选项。
3. 该方案无需修改原始JSON文件，也不需要引入第三方工具，仅通过代码预处理即可解决问题。

内容的提问来源于stack exchange，提问作者Prasad