将通用 avro 记录序列化为 Array[Byte] 将架构保留在对象中答案

【问题标题】：Serializing generic avro records as an Array[Byte] keeps the schema in the object将通用 avro 记录序列化为 Array[Byte] 将架构保留在对象中
【发布时间】：2018-08-01 13:32:57
【问题描述】：

情况

我目前正在使用 AVRO 和模式存储库编写消费者/生产者。

根据我收集到的信息，我对这些数据进行序列化的选项是使用 Confluent 的 avro 序列化程序，或者使用 Twitter 的 Bijection。

似乎双射看起来最直接。

所以我想以以下格式生成日期ProducerRecord[String,Array[Byte]]，这归结为 [一些字符串 ID，序列化的 GenericRecord]

（注意：我要使用通用记录，因为此代码库必须处理从 Json/csv/... 解析的数千个模式）

问题：

我序列化和使用 AVRO 的全部原因是您不需要在数据本身中包含架构（就像使用 Json/XML/...一样）。
然而，当检查主题中的数据时，我看到整个方案与数据一起包含。我做错了什么，这是设计使然，还是应该改用融合序列化程序？

代码：

  def jsonStringToAvro(jString: String, schema: Schema): GenericRecord = {
    val converter = new JsonAvroConverter
    val genericRecord = converter.convertToGenericDataRecord(jString.replaceAll("\\\\/","_").getBytes(), schema)

    genericRecord
  }
def serializeAsByteArray(avroRecord: GenericRecord): Array[Byte] = {
    //val genericRecordInjection = GenericAvroCodecs.toBinary(avroRecord.getSchema)
    val r: Array[Byte] = GenericAvroCodecs.toBinary(avroRecord.getSchema).apply(avroRecord)

    r
  }

//schema comes from a rest call to the schema repository
new ProducerRecord[String, Array[Byte]](topic, myStringKeyGoesHere, serializeAsByteArray(jsonStringToAvro(jsonObjectAsStringGoesHere, schema)))


        producer.send(producerRecord, new Callback {...})

【问题讨论】：

双射库不与模式注册表交互，并且您不会像 Confluent 序列化程序那样将 ID 放在任何地方。因此，整个架构将成为消息的一部分
另外，ProducerRecord[String, GenericRecord] 有什么问题？并将 REST 调用放入序列化程序中？
一些项目可能有很大的容量，所以我认为序列化会给我带来一些性能提升。那么，当我使用 Confluent 序列化程序时，这会从通用记录中剥离模式吗？
Kafka Serializer 接口旨在为您获取字节数组。在类的“主要方法”中编写该逻辑没有任何好处

标签： scala apache-kafka avro bijection

【解决方案1】：

如果您查看 Confluent source code ，您会看到与架构存储库交互的操作顺序是

从 Avro 记录中获取架构，并计算其 ID。理想情况下，将 Schema 发布到存储库，或者以其他方式散列它应该给你一个 ID。
分配一个字节缓冲区
将返回的ID写入缓冲区
将 Avro 对象值（不包括架构）作为字节写入缓冲区
将该字节缓冲区发送到 Kafka

目前，您的 Bijection 使用将在字节中包含架构，而不是用 ID 替换它

【讨论】：

谢谢，我想我做出了错误的假设，即 GenericRecord 只是需要读取外部架构的二进制 avro 数据。
GenericRecord 类似于 Hashmap。仍然有命名字段，所以它需要一些方法来解析它们并知道它们是否存在