如何在Avro Schema中引用其他Schema并在Python中正常解析

问题原因 #

默认avro.schema.parse方法只会解析你传入的单个Schema文本，不会自动扫描、加载依赖的外部Schema文件，所以主Schema中引用的其他自定义类型找不到定义，触发解析异常。

解决方案 #

方案1：手动加载所有依赖Schema批量解析 #

你需要先把所有关联的Schema文件全部读取出来，再通过make_avsc_object方法完成带依赖的解析，代码示例如下：

import avro.schema
import os
import json

schema_defs = {}
schema_dir = "./com.example.common"

# 遍历目录读取所有avsc文件
for filename in os.listdir(schema_dir):
    if filename.endswith(".avsc"):
        file_path = os.path.join(schema_dir, filename)
        with open(file_path, "r", encoding="utf-8") as f:
            schema_def = json.load(f)
            full_name = f"{schema_def['namespace']}.{schema_def['name']}"
            schema_defs[full_name] = schema_def

# 批量注册所有Schema到全局名称表
names = avro.schema.Names()
for schema_def in schema_defs.values():
    avro.schema.make_avsc_object(schema_def, names)

# 直接获取解析完成的主Schema
item_schema = names.get_name("com.example.common.Item")
print(item_schema)

方案2：合并为单Schema文件解析 #

也可以把所有Schema的JSON对象放到同一个数组文件中，avro解析时会自动处理同文件内的类型引用，合并后的文件结构示例：

[
    {
        "namespace":"com.example.common",
        "name":"ItemId",
        "type":"record",
        "fields":[{"name":"id","type":"int"}]
    },
    // 此处补充Range、Features的定义
    {...},
    {...},
    // 最后放置主Schema定义
    {
        "namespace":"com.example.common",
        "name":"Item",
        "type":"record",
        "fields":[
            {"name":"ItemId","type":"com.example.common.ItemId"},
            {"name":"features","type":"com.example.common.Features"}
        ]
    }
]

直接解析该合并文件即可：

import avro.schema

with open("./merged_schemas.avsc", "rb") as f:
    schema = avro.schema.parse(f.read())
print(schema)

注意事项 #

• 无论用哪种方案，都要确保所有被引用的Schema（包括间接引用的Range类型）都被完整加载，不能遗漏
• 不需要手动调整Schema的加载顺序，avro的解析逻辑会自动处理类型依赖

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