使用fastavro别名转换Schema时遭遇SchemaResolutionError求助

问题原因 #

触发SchemaResolutionError的核心原因是顶层记录类型不兼容。写入数据时用的old_schema记录名为generated，读取时指定的new_schema记录名为test，fastavro会校验顶层类型的匹配性——即使字段配置了别名，若顶层记录没有别名映射，会被判定为完全不同的类型，导致解析失败。

修复方案 #

在new_schema的顶层记录定义中添加aliases字段，将旧记录名generated纳入其中，让fastavro识别这两个记录类型是兼容的。修改后的new_schema代码如下：

new_schema = parse_schema({
    "type": "record",
    "namespace": "com.node40",
    "name": "test",
    "aliases": ["generated"],  # 新增：添加旧记录名作为顶层别名
    "fields": [
        {
            "name": "k1",
            "type": "string",
            "aliases": ["key1"]
        },
        {
            "name": "k2",
            "type": "string",
            "aliases": ["key2"]
        },
        {
            "name": "k3",
            "type": "string",
            "aliases": ["key3"]
        }
    ]
})

完整可运行代码 #

from fastavro import writer, reader, json_writer
from fastavro.schema import parse_schema
from io import BytesIO

# Sample data
input_json = [
    {
        "key1": "value1",
        "key2": "value2",
        "key3": "value3"
    }
]

# Old schema that matches the input_json
old_schema = parse_schema({
    "type": "record",
    "namespace": "com.node40",
    "name": "generated",
    "fields": [
        {
            "name": "key1",
            "type": "string"
        },
        {
            "name": "key2",
            "type": "string"
        },
        {
            "name": "key3",
            "type": "string"
        }
    ]
})

# New schema with old schema names as aliases (fixed version)
new_schema = parse_schema({
    "type": "record",
    "namespace": "com.node40",
    "name": "test",
    "aliases": ["generated"],  # 新增顶层记录别名
    "fields": [
        {
            "name": "k1",
            "type": "string",
            "aliases": ["key1"]
        },
        {
            "name": "k2",
            "type": "string",
            "aliases": ["key2"]
        },
        {
            "name": "k3",
            "type": "string",
            "aliases": ["key3"]
        }
    ]
})

# Write to buffer as serialized avro using old_schema
buffer = BytesIO()
writer(buffer, old_schema, input_json, validator=True)
buffer.seek(0)

# Read serialized avro from buffer, deserialize and write to json file
input_avro = reader(buffer, new_schema)
json_writer('fitted_data.json', new_schema, input_avro)

验证结果 #

运行修改后的代码，会成功生成fitted_data.json，文件内容为：

{"k1": "value1", "k2": "value2", "k3": "value3"}

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