PySpark无法读取Kafka流问题求助

环境版本 #

• kafka-python: 2.0.2
• Spark: 3.1.1

问题背景 #

使用kafka-python生产者向Kafka主题发送数据后，kafka-python消费者可正常读取所有历史消息，但PySpark结构化流读取该主题时无法获取数据。

排查及解决步骤 #

1. 确保kafka-python生产者的消息已被Kafka持久化

kafka-python的send()方法默认异步执行，若发送后程序直接退出，消息可能未完成写入Kafka。需等待Broker确认消息写入：

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='localhost:9092')
# 发送消息并等待Broker确认
send_future = producer.send(topic_name, str.encode(text+"\n"))
send_future.get(timeout=10)  # 等待10秒确保消息写入完成
producer.flush()  # 刷新缓存，确保所有消息发送至Kafka

2. 修正Spark Kafka读取配置并解析消息内容

• 简化bootstrap.servers配置，避免重复地址（如同时写127.0.0.1和localhost）导致解析异常
• Spark读取的Kafka消息默认是二进制格式，需解码为字符串才能在控制台看到可读内容

修改后的PySpark代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, decode

spark = SparkSession.builder.appName("test").getOrCreate()
# 读取Kafka流
kafka_df = spark.readStream.format("kafka") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("startingOffsets", "earliest") \
  .option("subscribe", topic_name) \
  .load()

# 解析二进制value为UTF-8字符串
parsed_df = kafka_df.select(
    col("topic"),
    col("partition"),
    col("offset"),
    decode(col("value"), "utf-8").alias("content")
)

# 输出到控制台
query = parsed_df.\
  writeStream.format("console").\
  trigger(processingTime='1 seconds').\
  option("checkpointLocation", "path/to/checkpoints").\
  start()

query.awaitTermination()

3. 重置Spark检查点偏移量

若之前运行过Spark流任务，checkpoint目录会保留已读取的偏移量，导致新任务不会重新读取历史数据：

• 删除指定的checkpointLocation目录后重新运行任务
• 或手动指定起始偏移量（以单分区主题为例）：

kafka_df = spark.readStream.format("kafka") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("startingOffsets", f"""{{"{topic_name}":{{"0":0}}}}""") \
  .option("subscribe", topic_name) \
  .load()

4. 验证网络连通性

确保Spark运行环境能访问Kafka Broker的9092端口：本地运行用localhost即可，分布式环境需使用Kafka节点的外部可访问IP/主机名。

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