Scala和Avro:将case类转换为avro记录

Question

我使用的是Scala 2.12和Avro (org.apache.avro) 1.8。

我有以下架构：

Schema: {"name": "person","type": "record","fields": [{"name": "address","type": {"type" : "record","name" : "AddressUSRecord","fields" : [{"name": "streetaddress", "type": "string"},{"name": "city", "type":"string"}]}}]}

对应的Scala case类包括：

case class AddressUSRecord (streetaddress: String, name: String}

case class Address (addressUSRecord: List[AddressUSRecord])

case class Person (person: Address)

我使用GenericRecord将case类PnlRecord的对象转换为Avro。

val schema = new Schema.Parser().parse(new File(schemaFileName))
val avroRecord = new GenericData.Record(schema)
val writer = new GenericDatumWriter[GenericRecord](schema)
val out = new ByteArrayOutputStream()
val encoder = EncoderFactory.get().binaryEncoder(out, null)
val producer = new KafkaProducer[String, Array[Byte]](properties)
avroRecord.put("header", record.header)
//Please note that this pnlData (see above case class) is complex and created accordingly.
avroRecord.put("pnlData", record.pnlData)
writer.write(avroRecord, encoder)
val bytes = out.toByteArray
encoder.flush()
out.close()

我得到了以下错误。

2019-03-13 21:57:29.832 [application-akka.actor.default-dispatcher-4] ERROR controllers.SAController.$anonfun$publishToSA$2(34) - ca.company.project.sa.model.MessageHeader cannot be cast to org.apache.avro.generic.IndexedRecord
java.lang.ClassCastException: ca.company.project.sa.model.MessageHeader cannot be cast to org.apache.avro.generic.IndexedRecord
        at org.apache.avro.generic.GenericData.getField(GenericData.java:697)
        at org.apache.avro.generic.GenericData.getField(GenericData.java:712)
        at org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:164)
        at org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:156)
        at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:118)
        at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
        at org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:166)
        at org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:156)
        at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:118)
        at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
        at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:62)

为什么我的MessageHeader case类不能强制转换为IndexedRecord？这里我漏掉了什么？

我们如何将这种复杂的案例类转换为avro对象？有人可以帮助这种嵌套的案例类例子转换为avro记录吗？

提前谢谢。

Answer 1

Confluent Kafka Avro序列化程序是基于Java的，因此很可能不是为使用Scala对象而设计的。我看到您的pnlBreakdown被声明为List[PnlBreakdown] -如果这是一个Scala列表，序列化程序甚至不会将其识别为一个集合。case类也是如此-如果没有@BeanProperty注释，这些类将不会被识别为Java Beans