我正在寻找一个示例来对 Avro SpecificRecordBase 对象进行双射，类似于 GenericRecordBase，或者是否有更简单的方法将 AvroSerializer 类用作 Kafka 键和值序列化程序。
Injection<GenericRecord, byte[]> genericRecordInjection =
GenericAvroCodecs.toBinary(schema);
byte[] bytes = genericRecordInjection.apply(type);
【问题讨论】：
标签： serialization apache-kafka avro bijection
https://github.com/miguno/kafka-storm-starter 提供了这样的示例代码。
例如，参见AvroDecoderBolt。来自它的 javadocs：
这个螺栓需要 Avro 编码的二进制格式的传入数据，根据
T的 Avro 模式进行序列化。它将传入的数据反序列化为Tpojo，并将此 pojo 发送给下游消费者。因此，这个螺栓可以被认为是 Twitter Bijection 的Injection.invert[T, Array[Byte]](bytes)用于 Avro 数据的 Storm 等价物。
在哪里
T：Avro 记录的类型（例如Tweet）基于正在使用的底层 Avro 模式。必须是 Avro 的SpecificRecordBase的子类。
代码的关键部分是（我把代码折叠成这个sn-p）：
// With T <: SpecificRecordBase
implicit val specificAvroBinaryInjection: Injection[T, Array[Byte]] =
SpecificAvroCodecs.toBinary[T]
val bytes: Array[Byte] = ...; // the Avro-encoded data
val decodeTry: Try[T] = Injection.invert(bytes)
decodeTry match {
case Success(pojo) =>
System.out.println("Binary data decoded into pojo: " + pojo)
case Failure(e) => log.error("Could not decode binary data: " + Throwables.getStackTraceAsString(e))
}
【讨论】：
Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(new File("/Users/.../schema.avsc"));
Injection<Command, byte[]> objectInjection = SpecificAvroCodecs.toBinary(schema);
byte[] bytes = objectInjection.apply(c);
【讨论】：