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

import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.apache.avro.io.{BinaryDecoder, BinaryEncoder, DecoderFactory, EncoderFactory}
import org.apache.avro.specific.{SpecificDatumReader, SpecificDatumWriter}

// 定义Avro Schema
val schemaString = """
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "int"}
  ]
}
"""
val schema = new Schema.Parser().parse(schemaString)

// 创建Avro对象
val user = new GenericRecordBuilder(schema)
  .set("name", "John")
  .set("age", 30)
  .build()

// 序列化为Avro二进制数据
val writer = new SpecificDatumWriter[GenericRecord](schema)
val out = new ByteArrayOutputStream()
val encoder: BinaryEncoder = EncoderFactory.get().binaryEncoder(out, null)
writer.write(user, encoder)
encoder.flush()
out.close()
val avroBytes = out.toByteArray()

// 发送Avro数据到Kafka
val producer = new KafkaProducer[String, Array[Byte]](props)
val record = new ProducerRecord[String, Array[Byte]]("topic", avroBytes)
producer.send(record)
producer.close()

from avro import schema, datafile, io

# 定义Avro Schema
schema_string = '''
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "int"}
  ]
}
'''
avro_schema = schema.Parse(schema_string)

# 创建Avro对象
user = {"name": "John", "age": 30}

# 序列化为Avro二进制数据
writer = io.DatumWriter(avro_schema)
bytes_writer = io.BytesIO()
encoder = io.BinaryEncoder(bytes_writer)
writer.write(user, encoder)
encoder.flush()
avro_bytes = bytes_writer.getvalue()

# 发送Avro数据到Kafka
producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer.send('topic', avro_bytes)
producer.close()