Kafka与PySpark集成程序无报错但查询返回空DataFrame如何解决

1. 核心逻辑问题修复 #

你当前的代码启动流作业后立即执行show()查询，此时Spark的微批处理还未完成Kafka数据拉取，内存表尚未写入任何数据，因此返回空结果。Structured Streaming是异步执行的，start()方法启动作业后不会阻塞主线程，需要添加等待逻辑让流有时间处理数据。
修复示例：

rawQuery = dsraw \
        .writeStream \
        .queryName("query1")\
        .format("memory")\
        .start()

# 等待3秒让流完成第一次数据拉取，长期运行可替换为rawQuery.awaitTermination()
import time
time.sleep(3)

raw = spark.sql("select * from query1")
raw.show()
# 测试完成后手动停止流作业避免资源占用
rawQuery.stop()

2. Kafka配置项排查 #

• 默认消费策略问题：Structured Streaming默认从最新偏移量开始消费，只会拉取Spark作业启动之后产生的消息，你提前生产的测试消息默认不会被读取。如果需要消费历史数据，需要添加如下配置：

dsraw = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "kafka:9092") \
.option("subscribe", "quickstart-events") \
.option("startingOffsets", "earliest") # 新增该行，从最早偏移量开始读取
.load()

• 连接地址验证：如果你是本地运行Spark、Kafka部署在容器内，先确认本地hosts已经配置kafka域名映射到正确IP，或者直接替换为Kafka实际可访问的IP+端口测试，比如127.0.0.1:9092。
• Kafka侧数据验证：通过官方控制台消费命令确认主题确实存在消息：
  kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic quickstart-events --from-beginning

3. 依赖包排查 #

确认启动PySpark作业时已经加载了对应版本的Kafka集成依赖，比如Spark3.3版本对应的启动参数为：
pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0
依赖的版本号需要和你实际使用的Spark版本、Scala版本完全匹配，缺少依赖时作业不会抛出异常，但也无法拉取到Kafka数据。

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