scala 和 Python 之间的 Avro Kafka 转换问题

Question

我们的项目有 scala 和 python 代码，我们需要 send/consume avro 编码消息到 kafka。

我正在使用 python 和 scala 向 kafka 发送 avro 编码消息。我在 scala 代码中有生产者，它使用 Twitter 双射库发送 avro 编码的消息，如下所示：

val resourcesPath = getClass.getResource("/avro/url_info_schema.avsc")
val schemaFile = scala.io.Source.fromURL(resourcesPath).mkString
val schema = parser.parse(schemaFile)
val recordInjection = GenericAvroCodecs[GenericRecord](schema)
val avroRecord = new GenericData.Record(schema)
avroRecord.put("url_sha256", row._1)
avroRecord.put("url", row._2._1)
avroRecord.put("timestamp", row._2._2)
val recordBytes = recordInjection.apply(avroRecord)
kafkaProducer.value.send("topic", recordBytes)

Avro 架构看起来像

{
  "namespace": "com.rm.avro",
  "type": "record",
  "name": "url_info",
  "fields":[
     {
        "name": "url_sha256", "type": "string"
     },
     {
        "name": "url",  "type": "string"
     },
     {
        "name": "timestamp", "type": ["long"]
     }
 ]

}

我能够在 scala 的 KafkaConsumer 中成功解码它

val resourcesPath = getClass.getResource("/avro/url_info_schema.avsc")
val schemaFile = scala.io.Source.fromURL(resourcesPath).mkString


kafkaInputStream.foreachRDD(kafkaRDD => {
  kafkaRDD.foreach(

    avroRecord => {
      val parser = new Schema.Parser()
      val schema = parser.parse(schemaFile)
      val recordInjection = GenericAvroCodecs[GenericRecord](schema)
      val record = recordInjection.invert(avroRecord.value()).get
      println(record)
    }
  )

}

但是，我无法解码 python 中的消息，出现以下异常

'utf8' codec can't decode byte 0xe4 in position 16: invalid continuation byte

python 代码如下所示： schema_path="avro/url_info_schema.avsc" 架构 = avro.schema.parse(打开(schema_path).read())

for msg in consumer:
   bytes_reader = io.BytesIO(msg.value)
    decoder = avro.io.BinaryDecoder(bytes_reader)
    reader = avro.io.DatumReader(schema)
    decoded_msg = reader.read(decoder)
    print(decoded_msg)

另外 python scala avro 消费者不理解 avro 生产者消息。我在那里得到一个例外。 Python Avro 制作人如下所示：

datum_writer = DatumWriter(schema)
bytes_writer = io.BytesIO()

datum_writer = avro.io.DatumWriter(schema)
encoder = avro.io.BinaryEncoder(bytes_writer)
datum_writer.write(data, encoder) 
raw_bytes = bytes_writer.getvalue()
producer.send(topic, raw_bytes)

如何在 python 和 scala 之间保持一致？任何指针都会很棒

Answer 1

我在 python 中使用二进制编码器，而在 Scala 中什么也没有。只需从

更改一行

val recordInjection = GenericAvroCodecs[GenericRecord](schema)

到

val recordInjection = GenericAvroCodecs.toBinary[GenericRecord](schema)

我希望其他人觉得它有用。 python 代码

无需更改

scala 和 Python 之间的 Avro Kafka 转换问题

Avro Kafka conversion issues between scala and Python

python

scala

apache-kafka

spark-avro