PySpark Structured Streaming读取Kafka无结果问题求助

我之前也碰到过一模一样的情况！既然你已经确认Kafka本身的收发机制没问题，那问题肯定出在PySpark结构化流的配置或者执行环节上，给你梳理几个最容易踩的坑和对应的解决办法：

1. 最常见的坑：忘记启动流的输出动作 #

PySpark的结构化流是惰性执行的——你只调用load()方法的话，Spark只会生成查询计划，根本不会真正去连接Kafka消费数据。必须通过writeStream定义输出目标并调用start()，才能触发流的运行。

比如你需要把数据输出到控制台查看的话，得加上这段代码：

# 先把Kafka的binary value转成字符串（不然直接打印是二进制内容）
parsed_events = events.selectExpr("CAST(value AS STRING) AS message")

# 启动流查询，输出到控制台
query = parsed_events.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

# 必须等待流运行，不然程序会直接退出
query.awaitTermination()

2. 检查消息的反序列化处理 #

Kafka默认发送的是二进制格式的key和value，你直接打印load()返回的DataFrame，看到的会是BinaryType的内容，看起来像是空的或者乱码。一定要把value（或者key）转成可读的格式，比如字符串：

# 用selectExpr直接转换，或者用functions.col+cast
from pyspark.sql.functions import col
parsed_events = events.withColumn("message", col("value").cast("string"))

3. startingOffsets配置是否符合预期 #

你用的是latest，这个配置表示Spark会从启动流之后产生的新消息开始消费。如果启动流之前Kafka主题里已经有消息，这些历史消息是不会被读取的。

如果想验证是否能读到数据，可以先改成earliest，让Spark消费主题里的所有历史消息：

.option("startingOffsets", "earliest")

4. 检查消费者组的偏移量状态 #

Spark会自动为Kafka流生成消费者组（如果没手动指定的话），可以用Kafka的命令行工具查看该组的偏移量是否已经追到最新：

# 先找到Spark用的消费者组名，一般是spark-kafka-source-开头的
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# 查看该组的偏移量详情
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group <你的消费者组名>

如果CURRENT-OFFSET等于LOG-END-OFFSET，说明已经没有新消息可以消费了，这时候往主题里发新消息就能看到输出。

5. 手动指定消费者组（可选） #

如果没指定group.id，每次重启Spark流都会生成新的消费者组，导致偏移量不保留。手动指定一个固定的组名可以让Spark重启后从上次的位置继续消费：

.option("group.id", "my-spark-kafka-consumer-group")

完整的测试代码 #

把上面的要点整合起来，你可以试试这段完整的代码：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("KafkaStreamTest").getOrCreate()

# 读取Kafka流
events = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "kafka_stream") \
    .option("startingOffsets", "earliest") \
    .option("group.id", "test-spark-group") \
    .load()

# 解析消息
parsed_events = events.selectExpr(
    "CAST(key AS STRING) AS key",
    "CAST(value AS STRING) AS message",
    "timestamp"
)

# 输出到控制台
query = parsed_events.writeStream \
    .outputMode("append") \
    .format("console") \
    .option("truncate", False) \
    .start()

# 保持程序运行
query.awaitTermination()

运行这段代码后，往kafka_stream主题发送几条消息，应该就能在控制台看到输出了。如果还是不行，去查看Spark的Driver日志，里面会有更详细的错误信息（比如Kafka连接超时、权限问题等）。

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