Spark Streaming Kafka Avro消费报错：'dict'对象无'split'属性

我一眼就看出问题出在哪了——你用Confluent反序列化器把Avro消息转成Python字典（dict）了，但原来的WordCount示例代码是给字符串消息写的，直接调用split()方法当然会报错，因为字典根本没有这个属性！

问题根源 #

PythonStreamingDirectKafkaWordCount示例默认假设从Kafka拿到的是纯字符串消息，所以直接对消息值调用split()拆分单词。但你用Avro反序列化后，每条消息都变成了一个字典（对应Avro的record结构），自然触发AttributeError: 'dict' object has no attribute 'split'。

解决方案 #

你需要先从反序列化后的字典里提取出包含文本内容的字段，再对这个字段的字符串值进行分词操作。下面是修改后的完整代码示例：

import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField

# 替换成你的Schema Registry地址和实际的Avro Schema
schema_registry_url = "http://your-schema-registry:8081"
schema_str = """
{
  "type": "record",
  "name": "TextRecord",
  "fields": [
    {"name": "content", "type": "string"},
    {"name": "timestamp", "type": "long"}
  ]
}
"""

# 初始化Schema Registry客户端
schema_registry_client = SchemaRegistryClient({"url": schema_registry_url})
avro_deserializer = AvroDeserializer(schema_registry_client, schema_str)

def deserialize_kafka_message(value):
    if value is None:
        return None
    # 把Avro二进制消息反序列化为Python字典
    return avro_deserializer(value, SerializationContext("your-kafka-topic", MessageField.VALUE))

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: PythonStreamingDirectKafkaAvroWordCount.py <broker_list> <topic>", file=sys.stderr)
        exit(-1)

    sc = SparkContext(appName="PythonStreamingDirectKafkaAvroWordCount")
    ssc = StreamingContext(sc, 2)  # 调整批处理间隔为2秒

    brokers, topic = sys.argv[1:]
    # 创建Kafka Direct Stream，使用自定义反序列化器
    kvs = KafkaUtils.createDirectStream(ssc, [topic],
                                        {"metadata.broker.list": brokers},
                                        valueDecoder=deserialize_kafka_message)

    # 关键步骤：从字典中提取文本字段（这里假设你的Avro消息有个叫content的字符串字段）
    # 用get方法避免KeyError，默认返回空字符串
    text_lines = kvs.map(lambda x: x[1].get("content", ""))

    # 现在text_lines是字符串类型的DStream，可以正常调用split分词了
    words = text_lines.flatMap(lambda line: line.split(" "))
    pairs = words.map(lambda word: (word, 1))
    wordCounts = pairs.reduceByKey(lambda x, y: x + y)

    wordCounts.pprint()

    ssc.start()
    ssc.awaitTermination()

几个要注意的点 #

• 一定要替换代码里的schema_registry_url、schema_str和Kafka主题名称，匹配你自己的环境。
• 确认你的Avro Schema里确实有用于分词的字符串字段（比如示例中的content），要换成你实际Schema中的字段名。
• 可以加一些额外的校验逻辑，比如判断反序列化后的结果是否是字典、目标字段是否存在且为字符串类型，避免运行时出现意外错误。
• 确保已经安装了依赖：pip install confluent-kafka pyspark，Spark版本要和集群兼容。

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