Kafka Avro反序列化触发'Type' KeyError错误求助

错误根源 #

你传给AvroDeserializer的schema_str参数值是'{"id": "4"}'，这不是合法的Avro Schema结构。fastavro解析时找不到Avro Schema必需的type字段，因此抛出KeyError。

解决方案 #

方案1：让Schema Registry自动拉取对应Schema（推荐） #

Confluent格式的Avro消息本身会携带Schema ID，AvroDeserializer可以通过Schema Registry客户端自动根据ID拉取对应的完整Schema，无需手动传入Schema字符串。

修改代码如下：

1. 调整反序列化函数：

def process_record_confluent(record: bytes, src: SchemaRegistryClient):
    # 无需传入schema_str，让Deserializer自动从Registry获取
    deserializer = AvroDeserializer(schema_registry_client=src)
    return deserializer(record, None)

2. 更新消费者的value_deserializer配置：

consumer = KafkaConsumer(
    'kafkamessages.dev.orders',
    bootstrap_servers=['localhost:9092'],
    group_id='payment_orders',
    auto_offset_reset='latest',
    # 移除schema参数
    value_deserializer=lambda m: process_record_confluent(m, src=SchemaRegistryClient({'url': 'https://kafka-schema.mysales.dep/'})),
    consumer_timeout_ms=6000
)

方案2：手动传入完整Avro Schema字符串 #

如果你需要指定固定的Reader Schema，必须传入完整的Avro Schema JSON字符串，而不是仅Schema ID。

比如将你的完整Schema整理成合法JSON后传入：

# 完整的Avro Schema字符串
full_avro_schema = '''{
  "type": "record",
  "name": "Envelope",
  "namespace": "xxxxxx",
  "fields": [
    // 补充你的字段定义
  ]
}'''

def process_record_confluent(record: bytes, src: SchemaRegistryClient, schema: str):
    deserializer = AvroDeserializer(schema_str=schema, schema_registry_client=src)
    return deserializer(record, None)

# 消费者配置中传入完整Schema
consumer = KafkaConsumer(
    'kafkamessages.dev.orders',
    bootstrap_servers=['localhost:9092'],
    group_id='payment_orders',
    auto_offset_reset='latest',
    value_deserializer=lambda m: process_record_confluent(m, src=SchemaRegistryClient({'url': 'https://kafka-schema.mysales.dep/'}), schema=full_avro_schema),
    consumer_timeout_ms=6000
)

额外注意事项 #

• 确保Schema Registry客户端配置正确（如URL、认证信息等），否则会出现无法拉取Schema的错误。
• 确认Kafka中的消息是Confluent标准的Avro格式（开头包含1字节magic值+4字节Schema ID），否则AvroDeserializer无法正确解析。

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