Python 3实现无Schema多Avro序列化对象写入文件方法咨询

我之前也碰到过类似需求，用fastavro库就能完美解决——它比官方的avro库更轻量高效，而且直接支持单个对象的序列化/反序列化操作，刚好符合你不需要写入Schema的要求。下面分写入和读取两部分给你具体代码：

第一步：安装依赖 #

先确保你装了fastavro：

pip install fastavro

第二步：写入多个Avro对象（不含Schema） #

假设你已经有了预先定义好的Schema（比如从JSON文件加载，或者直接在代码里定义），我们直接序列化每个对象的二进制数据并写入文件：

from fastavro import serialize
import json

# 1. 加载你的预先定义的Schema（这里示例直接写，实际可以从文件读取）
schema_json = {
    "type": "record",
    "name": "User",
    "fields": [
        {"name": "id", "type": "int"},
        {"name": "name", "type": "string"},
        {"name": "email", "type": ["null", "string"], "default": None}
    ]
}

# 2. 准备要写入的多个对象
objects_to_write = [
    {"id": 1, "name": "Alice", "email": "alice@example.com"},
    {"id": 2, "name": "Bob"},
    {"id": 3, "name": "Charlie", "email": "charlie@example.com"}
]

# 3. 写入文件（二进制模式）
with open("avro_objects.bin", "wb") as f:
    for obj in objects_to_write:
        # 序列化单个对象为字节，直接写入文件
        serialize(f, obj, schema_json)

这里关键是用fastavro.serialize而不是fastavro.writer——后者会自动写入Avro文件的头信息（包括Schema），而前者只输出单个对象的二进制编码，完全符合你的需求。

第三步：用预先Schema读取文件 #

读取的时候，同样用fastavro的deserialize函数，循环从文件流中逐个解析对象：

from fastavro import deserialize
import json

# 1. 加载和写入时完全相同的Schema
schema_json = {
    "type": "record",
    "name": "User",
    "fields": [
        {"name": "id", "type": "int"},
        {"name": "name", "type": "string"},
        {"name": "email", "type": ["null", "string"], "default": None}
    ]
}

# 2. 读取文件并逐个反序列化
with open("avro_objects.bin", "rb") as f:
    while True:
        try:
            # 从文件流中读取并反序列化一个对象
            obj = deserialize(f, schema_json)
            print(obj)
        except EOFError:
            # 读到文件末尾时退出循环
            break

关键注意事项 #

• Schema一致性：写入和读取必须使用完全相同的Schema，否则反序列化会失败（Avro对Schema匹配要求严格）。
• 二进制模式：文件必须以二进制模式打开（wb/rb），否则会破坏Avro的二进制编码。
• 如果用官方avro库的替代方案：
  如果你一定要用官方的avro库，也可以通过BinaryEncoder和BinaryDecoder实现：

  # 写入（官方库）
  from avro.io import DatumWriter, BinaryEncoder
  import avro.schema
  
  schema = avro.schema.parse(json.dumps(schema_json))
  with open("avro_objects.bin", "wb") as f:
      writer = DatumWriter(schema)
      encoder = BinaryEncoder(f)
      for obj in objects_to_write:
          writer.write(obj, encoder)
  
  # 读取（官方库）
  from avro.io import DatumReader, BinaryDecoder
  
  schema = avro.schema.parse(json.dumps(schema_json))
  with open("avro_objects.bin", "rb") as f:
      reader = DatumReader(schema)
      decoder = BinaryDecoder(f)
      while True:
          try:
              obj = reader.read(decoder)
              print(obj)
          except EOFError:
              break
  

  不过官方库的性能比fastavro差不少，所以更推荐前者。

内容的提问来源于stack exchange，提问作者Daedalus