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": "missing@example.com", "doc": "email address"}
     ]
}
""")

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

如果我用schema_v1写一个avro记录，根据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

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": "missing@example.com",
            "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': 'missing@example.com'}

Answer 2

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