使用Confluent-Kafka Python实现流表Join时遇TypeError错误求助

这个TypeError: You must pass either str or Schema错误的核心原因是JSONDeserializer初始化时传入了非法的schema参数——你试图直接传递自定义的LookupData类，但JSONDeserializer只接受两种合法参数：字符串格式的JSON Schema，或者符合Confluent规范的Schema对象。

修复步骤 #

1. 定义匹配数据结构的JSON Schema字符串

根据你提供的生产者示例数据，定义对应的JSON Schema：

{
  "type": "object",
  "properties": {
    "id": {"type": "integer"},
    "name": {"type": "string"}
  },
  "required": ["id"]
}

2. 正确初始化JSONDeserializer

不要直接把LookupData类传给deserializer，而是通过from_dict回调将反序列化后的字典转换为自定义类对象：

class LookupData:
    def __init__(self, id, name):
        self.id = id
        self.name = name

    @classmethod
    def from_dict(cls, data):
        # 处理可选字段，避免KeyError
        return cls(id=data["id"], name=data.get("name"))

# 导入依赖
from confluent_kafka.schema_registry.json_schema import JSONDeserializer

# 初始化JSONDeserializer，传入合法的schema字符串
lookup_schema_str = """
{
  "type": "object",
  "properties": {
    "id": {"type": "integer"},
    "name": {"type": "string"}
  },
  "required": ["id"]
}
"""

json_deserializer = JSONDeserializer(
    schema_str=lookup_schema_str,
    from_dict=LookupData.from_dict  # 自动将字典转成LookupData实例
)

3. 在流/表操作中正确配置反序列化

确保创建KStream或KTable时，指定value_deserializer为上述初始化好的json_deserializer，示例代码：

from confluent_kafka import Consumer
from confluent_kafka.serialization import SerializationContext, MessageField

def consume_lookup_data(consumer, topic):
    consumer.subscribe([topic])
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            print(f"Consumer error: {msg.error()}")
            continue
        
        # 反序列化并转换为LookupData对象
        lookup_data = json_deserializer(
            msg.value(),
            SerializationContext(msg.topic(), MessageField.VALUE)
        )
        print(f"Received lookup data: ID={lookup_data.id}, Name={lookup_data.name}")

额外排查点 #

• 确认没有误将LookupData类直接传给JSONDeserializer的schema参数（这是最常见的错误）
• 如果使用Schema Registry托管schema，需通过SchemaRegistryClient获取合法的Schema对象，而非自定义类
• 验证生产者发送的消息是严格的JSON格式，无语法错误

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