如何用Python基于MongoDB JSON自动生成Avro Schema？

可以通过「业务JSON → JSON Schema → Avro Schema」的流程实现自动生成，这是处理此类场景的成熟方案。先解释你之前遇到的错误，再一步步说明具体实现：

为什么avro.schema.parse会报错？ #

avro.schema.parse()方法的作用是解析Avro Schema的JSON定义，而不是直接解析业务JSON数据。你传入的是MongoDB导出的业务JSON，缺少Avro Schema必须的type等关键字段，所以必然抛出SchemaParseException: No "type" property错误。

步骤1：从多条JSON样本生成JSON Schema #

单条JSON可能存在字段缺失、类型不完整的情况，建议用多条MongoDB导出的JSON样本生成更准确的JSON Schema。可以用Python的genson库自动推导：

1. 安装依赖：

pip install genson

2. 生成JSON Schema的代码：

import json
from genson import SchemaBuilder

# 加载多条JSON样本（每行一条MongoDB导出的JSON）
samples = []
with open('mongo_samples.json', 'r') as f:
    for line in f:
        samples.append(json.loads(line.strip()))

# 用样本构建JSON Schema
builder = SchemaBuilder()
for sample in samples:
    builder.add_object(sample)

# 导出JSON Schema
json_schema = builder.to_schema()
with open('parcel_json_schema.json', 'w') as f:
    json.dump(json_schema, f, indent=2)

这个工具会自动识别：

• 字段的基础类型（字符串、数字、布尔值等）
• 嵌套对象、数组结构
• 可空字段（包含null值的字段会生成anyOf规则）

步骤2：将JSON Schema转换为Avro Schema #

用avro-json-schema库可以直接把JSON Schema转换成符合规范的Avro Schema：

1. 安装依赖：

pip install avro-json-schema

2. 转换代码：

import json
from avro_json_schema import to_avro_schema

# 加载之前生成的JSON Schema
with open('parcel_json_schema.json', 'r') as f:
    json_schema = json.load(f)

# 转换为Avro Schema
avro_schema = to_avro_schema(json_schema)

# 导出Avro Schema（.avsc是Avro Schema的标准后缀）
with open('parcel_avro_schema.avsc', 'w') as f:
    json.dump(avro_schema, f, indent=2)

关键转换规则说明： #

• JSON Schema的string类型 → Avro的string类型（日期时间字符串会保留为string，Avro无原生datetime类型，通常用ISO格式字符串存储）
• JSON Schema的integer/number → Avro的int/float
• 嵌套对象 → Avro的record类型
• 数组 → Avro的array类型
• 可空字段（anyOf: ["null", "type"]）→ Avro的联合类型（如["null", "string"]）

后续优化建议 #

1. 手动调整Schema：生成的Avro Schema可以补充命名空间、字段描述，或者把重复的嵌套结构（比如route_parts里的终端信息）提取为可复用的record类型，提升Schema的可读性和可维护性。
2. 处理MongoDB ObjectId：如果_id是MongoDB的ObjectId，生成的Schema会是string类型；如果需要保留二进制格式，可以手动把_id的类型改成bytes，但要提前处理ObjectId的序列化逻辑。
3. 验证Schema合法性：用avro.schema.parse()验证生成的Avro Schema，再用样本数据测试序列化/反序列化，确保没有问题：

import avro.schema
from avro.datafile import DataFileReader, DataFileWriter
from avro.io import DatumReader, DatumWriter

# 验证Schema
schema = avro.schema.parse(json.dumps(avro_schema))
# 测试序列化
with open('test.avro', 'wb') as f:
    writer = DataFileWriter(f, DatumWriter(), schema)
    writer.append(samples[0])
    writer.close()

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