使用Confluent Kafka Python构建Avro生产者遇序列化错误求助

错误信息 #

Traceback (most recent call last):
  File "fastavro\_write.pyx", line 363, in fastavro._write.write_record
**TypeError: Expected dict, got Address**

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:\Users\deepa\KafkaFastApi\kafka\producer_avro.py", line 102, in <module>
    value=avro_serializer(user, SerializationContext(topic, MessageField.VALUE)),
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\deepa\KafkaFastApi\venv\Lib\site-packages\confluent_kafka\schema_registry\avro.py", line 304, in __call__
    schemaless_writer(fo, self._parsed_schema, value)
  File "fastavro\_write.pyx", line 794, in fastavro._write.schemaless_writer
  File "fastavro\_write.pyx", line 458, in fastavro._write.write_data
  File "fastavro\_write.pyx", line 402, in fastavro._write.write_record
  File "fastavro\_write.pyx", line 458, in fastavro._write.write_data
  File "fastavro\_write.pyx", line 378, in fastavro._write.write_record
**ValueError: no value and no default for address_line1**

问题原因 #

1. TypeError: Expected dict, got Address：user_to_dict函数中直接返回了Pydantic的Address实例，fastavro仅支持序列化字典类型，无法处理Pydantic对象，因此抛出类型错误。
2. ValueError: no value and no default for address_line1：这是第一个错误引发的连锁异常。fastavro无法解析Address对象，导致无法识别其中的address_line1字段，误判该字段未赋值且无默认值。
3. 额外潜在问题：Avro Schema中address_line2定义为必填string类型，但Pydantic模型中该字段是可选值（Optional[str] = None），若实际赋值为None，后续仍会触发序列化错误。

解决方案 #

1. 修正user_to_dict函数，将Pydantic对象转为字典 #

把Address实例通过Pydantic方法转为字典，同时将枚举类型转换为对应字符串（匹配Avro枚举要求）：

def user_to_dict(user: Person, ctx):
    """将Person实例转为Avro可序列化的字典"""
    # 转换Address实例为字典，枚举转字符串
    address_dict = user.address.dict()  # Pydantic v2请用model_dump()
    address_dict["address_type"] = address_dict["address_type"].value
    
    return dict(
        id=user.id,
        first_name=user.first_name,
        middle_name=user.middle_name,
        last_name=user.last_name,
        dob=user.dob,
        address=address_dict
    )

2. 对齐Avro Schema与Pydantic模型 #

修改Avro Schema中address_line2的类型为可空字符串，与Pydantic定义保持一致：

{
  "name": "address_line2",
  "type": ["string", "null"]
}

3. 确保枚举类型匹配 #

保证AddressType枚举的符号与Avro Schema中的event_type_enums完全一致（大小写敏感）：

from enum import StrEnum

class AddressType(StrEnum):
    mailing = "Mailing"
    residential = "Residential"

修改后的核心代码片段 #

# ... 其他导入与配置代码 ...

def user_to_dict(user: Person, ctx):
    address_dict = user.address.dict()
    address_dict["address_type"] = address_dict["address_type"].value
    return dict(
        id=user.id,
        first_name=user.first_name,
        middle_name=user.middle_name,
        last_name=user.last_name,
        dob=user.dob,
        address=address_dict
    )

avro_serializer = AvroSerializer(schema_registry_client, schema_str, user_to_dict)

# ... 消息生产代码 ...

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