avro-python3不提供模式演化？

Question

我尝试使用avro-python3（向后兼容性）重新创建一个模式演变案例。

我有两个模式：

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

schema_v1 = avro.schema.Parse("""
{
     "type": "record",
     "namespace": "com.example",
     "name": "CustomerV1",
     "fields": [
       { "name": "first_name", "type": "string", "doc": "First Name of Customer" },
       { "name": "last_name", "type": "string", "doc": "Last Name of Customer" },
       { "name": "age", "type": "int", "doc": "Age at the time of registration" },
       { "name": "height", "type": "float", "doc": "Height at the time of registration in cm" },
       { "name": "weight", "type": "float", "doc": "Weight at the time of registration in kg" },
       { "name": "automated_email", "type": "boolean", "default": true, "doc": "Field indicating if the user is enrolled in marketing emails" }
     ]
}
""")

schema_v2 = avro.schema.Parse("""
{
     "type": "record",
     "namespace": "com.example",
     "name": "CustomerV2",
     "fields": [
       { "name": "first_name", "type": "string", "doc": "First Name of Customer" },
       { "name": "last_name", "type": "string", "doc": "Last Name of Customer" },
       { "name": "age", "type": "int", "doc": "Age at the time of registration" },
       { "name": "height", "type": "float", "doc": "Height at the time of registration in cm" },
       { "name": "weight", "type": "float", "doc": "Weight at the time of registration in kg" },
       { "name": "phone_number", "type": ["null", "string"], "default": null, "doc": "optional phone number"},
       { "name": "email", "type": "string", "default": "[e-mail]", "doc": "email address"}
     ]
}
""")

第二个模式没有automated_email字段，但有两个附加字段：phone_number和email。

根据avro模式演化规则，如果我用schema_v1写入avro记录：

writer = DataFileWriter(open("customer_v1.avro", "wb"), DatumWriter(), schema_v1)
writer.append({
    "first_name": "John",
    "last_name": "Doe",
    "age" : 34, 
    "height": 178.0,
    "weight": 75.0,
    "automated_email": True
})
writer.close()

…我可以使用schema_v2读取它，前提是不存在字段有默认值

reader = DataFileReader(open("customer_v1.avro", "rb"), DatumReader(reader_schema=schema_v2))

for field in reader:
    print(field)

reader.close()

但我得到了以下错误：

SchemaResolutionException: Schemas do not match.

我知道这在Java中有效。这是一个视频课程的示例。有没有办法让它在python中工作？

Answer 1

如果您将第二个模式从CustomerV2更改为CustomerV1，它适用于avro-python3版本1.10.0。

Answer 2

fastavro，另一种python实现，可以很好地处理这个问题。

用第一个模式编写的代码如下:

s1 = {
    "type": "record",
    "namespace": "com.example",
    "name": "CustomerV1",
    "fields": [
        {"name": "first_name", "type": "string", "doc": "First Name of Customer"},
        {"name": "last_name", "type": "string", "doc": "Last Name of Customer"},
        {"name": "age", "type": "int", "doc": "Age at the time of registration"},
        {
            "name": "height",
            "type": "float",
            "doc": "Height at the time of registration in cm",
        },
        {
            "name": "weight",
            "type": "float",
            "doc": "Weight at the time of registration in kg",
        },
        {
            "name": "automated_email",
            "type": "boolean",
            "default": True,
            "doc": "Field indicating if the user is enrolled in marketing emails",
        },
    ],
}

record = {
    "first_name": "John",
    "last_name": "Doe",
    "age": 34,
    "height": 178.0,
    "weight": 75.0,
    "automated_email": True,
}

import fastavro

with open("test.avro", "wb") as fp:
    fastavro.writer(fp, fastavro.parse_schema(s1), [record])

用第二种模式阅读:

s2 = {
    "type": "record",
    "namespace": "com.example",
    "name": "CustomerV2",
    "fields": [
        {"name": "first_name", "type": "string", "doc": "First Name of Customer"},
        {"name": "last_name", "type": "string", "doc": "Last Name of Customer"},
        {"name": "age", "type": "int", "doc": "Age at the time of registration"},
        {
            "name": "height",
            "type": "float",
            "doc": "Height at the time of registration in cm",
        },
        {
            "name": "weight",
            "type": "float",
            "doc": "Weight at the time of registration in kg",
        },
        {
            "name": "phone_number",
            "type": ["null", "string"],
            "default": None,
            "doc": "optional phone number",
        },
        {
            "name": "email",
            "type": "string",
            "default": "[e-mail]",
            "doc": "email address",
        },
    ],
}

import fastavro

with open("test.avro", "rb") as fp:
    for record in fastavro.reader(fp, fastavro.parse_schema(s2)):
        print(record)

输出为预期的新字段：

{'first_name': 'John', 'last_name': 'Doe', 'age': 34, 'height': 178.0, 'weight': 75.0, 'phone_number': None, 'email': '[e-mail]'}