如何在Python中正确处理含联合类型的Avro JSON转Avro文件？

核心问题原因 #

Java的avro-tools的fromjson命令支持一种扩展的联合类型JSON编码格式（允许用{"类型名": 值}的结构指定联合类型的具体类型），但Python的avro库（包括avro-python3和fastavro）严格遵循Avro原生数据规范，不需要这种类型包装结构，直接传入对应类型的原生值即可。你的输入数据{"field1": {"int": 17} }是Java工具的扩展格式，不符合Python库的要求，因此报错。

正确处理方式 #

方式1：修改输入JSON为原生格式 #

对于联合类型["null", "int"]，正确的JSON数据格式应该是：

• 当取值为int时：{"field1": 17}
• 当取值为null时：{"field1": null}

用avro-python3实现的示例代码

import avro.schema
from avro.datafile import DataFileWriter
from avro.io import DatumWriter

# 解析Schema
schema = avro.schema.parse('''
{"type": "record", "name": "ROOT", "fields": [{"name": "field1", "type": ["null","int"] }] }
''')

# 符合规范的原生数据
valid_data = {"field1": 17}

# 写入Avro文件
with open("sample1.avro", "wb") as f:
    writer = DataFileWriter(f, DatumWriter(), schema)
    writer.append(valid_data)
    writer.close()

用fastavro实现的示例代码

from fastavro import writer, parse_schema

# 解析Schema
schema = parse_schema({
    "type": "record", "name": "ROOT", "fields": [{"name": "field1", "type": ["null","int"] }]
})

# 符合规范的原生数据
valid_data = {"field1": 17}

# 写入Avro文件
with open("sample1_fastavro.avro", "wb") as f:
    writer(f, schema, [valid_data])

方式2：手动转换带类型包装的输入数据 #

如果无法修改原始JSON格式，需要编写转换函数，将{"int": 17}、{"null": null}这类包装结构转换为原生值：

def convert_union_wrapped_data(data):
    if isinstance(data, dict):
        # 处理联合类型的单键包装结构
        if len(data) == 1:
            type_key, value = next(iter(data.items()))
            # 覆盖常见的Avro类型，可根据需求扩展
            if type_key == "null":
                return None
            elif type_key in ["int", "long", "float", "double", "string", "boolean"]:
                return value
        # 递归处理嵌套的字典结构
        return {k: convert_union_wrapped_data(v) for k, v in data.items()}
    elif isinstance(data, list):
        # 递归处理列表中的每个元素
        return [convert_union_wrapped_data(item) for item in data]
    else:
        # 非容器类型直接返回
        return data

# 原始带类型包装的数据
raw_data = {"field1": {"int": 17}}
# 转换为符合规范的数据
converted_data = convert_union_wrapped_data(raw_data)

# 之后即可用converted_data调用avro-python3或fastavro的写入逻辑

内容的提问来源于stack exchange，提问作者Piotr Śniady