如何在Java中动态解析Avro SpecificRecord并转换为KV结构？

背景与需求 #

刚接触Avro和Beam，需要编写一个PTransform，将Kafka消费来的SpecificRecord转换为类似Map结构的ElasticRow。目标是遍历SpecificRecord的所有嵌套字段，将键值对写入ElasticRow，且该PTransform要支持处理多个Kafka主题的不同Avro Schema。

现有代码与尝试思路 #

已编写的PTransform片段 #

private static final class FlattenSpecificRecordRecord
      implements SerializableFunction<KV<String, SpecificRecord>, Iterable<ElasticRow>> {
    private final Stream _stream;

    FlattenSpecificRecordRecord(Stream stream) {
      this._stream = stream;
    }

    public static FlattenSpecificRecordRecord of(Stream stream) {
      return new FlattenSpecificRecordRecord(stream);
    }

    @Override
    public Iterable<ElasticRow> apply(KV<String, SpecificRecord> recordKV) {
      // ElasticRow is like a map
      SpecificRecord record = recordKV.getValue();
      // How to traverse this?
      return ???
    }
  }

尝试的字段遍历实现 #

public class RecordConverter implements Serializable {
  private final String _className;
  private final SpecificRecord _record;

  public RecordConverter(String className, SpecificRecord record) {
    this._className = className;
    this._record = record;
  }

  public List<Object> execute() {
    List<Object> objects = new ArrayList<>();
    Schema schema = _record.getSchema();
    GenericRecord avroRecord = new GenericData.Record(_record.getSchema());
    Queue<Schema.Field> fields = new LinkedList<>();
    if (schema.getType() == Schema.Type.RECORD) {
      fields.addAll(schema.getFields());
    }
    for (Schema.Field field : fields) {
      if (field.schema().getType() == Schema.Type.RECORD) {
        fields.add(field);
        continue;
      }
      Object obj = avroRecord.get(field.name());
      System.out.println("This is your value " + obj);
      objects.add(obj);
    }
    return objects;
  }
}

示例嵌套Avro Schema #

{
  "type": "record",
  "name": "Address",
  "fields": [
    {
      "name": "streetaddress",
      "type": "string"
    },
    {
      "name": "city",
      "type": "City"
    }
  ]
}

当前疑问 #

• 现有遍历方案是否可行？
• 是否需要使用DatumReader来处理SpecificRecord？
• 如何利用类名（如com.company.MyAvroObject）通过反射实现通用的嵌套字段转换？

可行实现方案 #

1. 通用嵌套字段扁平化工具类 #

利用Avro自身的Schema和SpecificRecordAPI，通过递归遍历所有嵌套字段，生成扁平的键值对（嵌套路径用点分隔，比如address.city.name），无需反射或额外的DatumReader：

public class AvroFlattener implements Serializable {

    // 扁平化嵌套SpecificRecord，返回扁平键值对Map
    public static Map<String, Object> flatten(SpecificRecord record) {
        Map<String, Object> flatMap = new HashMap<>();
        flattenRecord(record, "", flatMap);
        return flatMap;
    }

    private static void flattenRecord(SpecificRecord record, String parentPath, Map<String, Object> flatMap) {
        Schema schema = record.getSchema();
        for (Schema.Field field : schema.getFields()) {
            String fieldName = field.name();
            String currentPath = parentPath.isEmpty() ? fieldName : parentPath + "." + fieldName;
            Object value = record.get(fieldName);

            if (value instanceof SpecificRecord) {
                // 遇到嵌套Record，递归处理
                flattenRecord((SpecificRecord) value, currentPath, flatMap);
            } else if (value instanceof Collection) {
                // 处理集合类型，可按需调整格式，这里转成字符串示例
                flatMap.put(currentPath, value.toString());
            } else {
                // 基础类型直接存入Map
                flatMap.put(currentPath, value);
            }
        }
    }
}

2. 整合到Beam的PTransform中 #

修改原有FlattenSpecificRecordRecord类，调用工具类生成扁平Map，再转换为ElasticRow：

private static final class FlattenSpecificRecordRecord
      implements SerializableFunction<KV<String, SpecificRecord>, Iterable<ElasticRow>> {
    private final Stream _stream;

    FlattenSpecificRecordRecord(Stream stream) {
      this._stream = stream;
    }

    public static FlattenSpecificRecordRecord of(Stream stream) {
      return new FlattenSpecificRecordRecord(stream);
    }

    @Override
    public Iterable<ElasticRow> apply(KV<String, SpecificRecord> recordKV) {
      SpecificRecord record = recordKV.getValue();
      // 扁平化嵌套字段
      Map<String, Object> flatMap = AvroFlattener.flatten(record);
      
      // 将扁平Map转换为ElasticRow（假设ElasticRow支持put方法）
      ElasticRow row = new ElasticRow();
      for (Map.Entry<String, Object> entry : flatMap.entrySet()) {
          row.put(entry.getKey(), entry.getValue());
      }
      
      return Collections.singletonList(row);
    }
  }

3. 关键问题说明 #

• 关于DatumReader：DatumReader的作用是从二进制Avro数据反序列化为对象，你已经拿到了SpecificRecord实例，不需要再用它读取数据，直接使用实例的API即可。
• 关于反射：不需要反射来遍历字段，Avro的Schema已经包含了所有字段信息，结合SpecificRecord.get(fieldName)就能安全取值，比反射更可靠且性能更好。如果需要动态创建SpecificRecord实例，才需要用到反射，但当前场景不需要。
• 原有思路的问题：你之前的RecordConverter里创建GenericData.Record是多余的，直接用传入的SpecificRecord实例即可；队列处理嵌套会导致死循环（重复添加同一个字段），递归是更合理的嵌套遍历方式。

内容的提问来源于stack exchange，提问作者Frederick Álvarez