Como converter avro aninhado GenericRecord para Row
Eu tenho um código para converter meu registro avro para Row usando a funçãoavroToRowConverter()
directKafkaStream.foreachRDD(rdd -> {
JavaRDD<Row> newRDD= rdd.map(x->{
Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(SchemaRegstryClient.getLatestSchema("poc2"));
return avroToRowConverter(recordInjection.invert(x._2).get());
});
Esta função não está funcionando para o esquema aninhado(TYPE= UNION)
.
private static Row avroToRowConverter(GenericRecord avroRecord) {
if (null == avroRecord) {
return null;
}
//GenericData
Object[] objectArray = new Object[avroRecord.getSchema().getFields().size()];
StructType structType = (StructType) SchemaConverters.toSqlType(avroRecord.getSchema()).dataType();
for (Schema.Field field : avroRecord.getSchema().getFields()) {
if(field.schema().getType().toString().equalsIgnoreCase("STRING") || field.schema().getType().toString().equalsIgnoreCase("ENUM")){
objectArray[field.pos()] = ""+avroRecord.get(field.pos());
}else {
objectArray[field.pos()] = avroRecord.get(field.pos());
}
}
return new GenericRowWithSchema(objectArray, structType);
}
Alguém pode sugerir como posso converter esquemas complexos em ROW?