使用PySpark的KafkaUtils.createDirectStream时遇AttributeError错误

这个错误我之前也碰到过，本质是PySpark无法识别你传入的fromOffsets中的分区对象，导致无法转换为底层Java API需要的格式。下面给你拆解问题和解决方案：

错误原因 #

你遇到的AttributeError: 'TopicPartition' object has no attribute '_jTopicAndPartition'，是因为传入fromOffsets的键不是PySpark Streaming Kafka API提供的正确的TopicPartition实例。PySpark的TopicPartition类封装了Java侧的对象，_jTopicAndPartition是它用来关联Java对象的内部属性，如果你的分区对象不是这个类的实例（比如误用了kafka-python库的TopicPartition，或者自己手动构造了类似对象），就会触发这个错误。

解决方案 #

1. 导入正确的TopicPartition类 #

确保你从PySpark的Kafka Streaming模块中导入，而非其他第三方库：

# Spark 1.x 版本（对应Kafka 0.8/0.9）
from pyspark.streaming.kafka import TopicPartition
# Spark 2.x+ 版本（对应Kafka 0.10+）
from pyspark.streaming.kafka010 import TopicPartition

2. 正确构造fromOffsets字典 #

用上述TopicPartition实例作为键，对应的值是你要指定的起始offset（整数类型）：

# 示例：指定test主题的0号分区从offset 100开始消费
fromOffset = {TopicPartition("test", 0): 100}

3. 匹配Spark与Kafka版本的API参数 #

不同版本的Spark和Kafka对应不同的API参数：

• Spark 1.x：kafkaParams用metadata.broker.list指定broker地址
• Spark 2.x+：kafkaParams用bootstrap.servers代替，且需导入kafka010下的模块

完整可运行示例（Spark 1.x版本） #

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils, TopicPartition

# 初始化上下文
sc = SparkContext(appName="KafkaDirectOffsetExample")
ssc = StreamingContext(sc, 5)  # 5秒批处理间隔

brokers = "your-broker-ip:9092"
topics = ["your-topic-name"]

# 构造指定offset的字典
fromOffset = {TopicPartition(topics[0], 0): 100}  # 消费第0个分区，从offset100开始

def messageHandler(message):
    # 自定义消息处理逻辑，这里返回消息内容
    return message[1]

# 创建DirectStream
stream = KafkaUtils.createDirectStream(
    ssc=ssc,
    topics=topics,
    kafkaParams={"metadata.broker.list": brokers},
    fromOffsets=fromOffset,
    messageHandler=messageHandler
)

# 打印处理结果
stream.pprint()

# 启动流处理
ssc.start()
ssc.awaitTermination()

如果是从checkpoint恢复offset，要确保checkpoint中保存的fromOffsets格式正确，或者在恢复时重新用正确的TopicPartition实例构造字典。

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