PySpark结构化流读取Kafka无数据，普通读取正常的原因排查

你遇到的问题是PySpark结构化流无法读取Kafka数据，但批处理模式可以正常读取，可能的原因如下：

1. 消费起始位置默认值导致错过历史数据 #

结构化流读取Kafka时，默认从latest偏移量开始消费，也就是只处理流启动之后新写入Kafka的消息。如果你的stations-topic里只有历史数据，流启动后没有新消息产生，自然不会有输出。而批处理的read模式默认会读取主题内所有历史数据。

解决方法：在流读取配置中添加起始偏移量参数：

df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", KAFKA_SERVER) \
    .option("subscribe", KAFKA_TOPIC) \
    .option("startingOffsets", "earliest")  # 从最开始的位置消费
    .load()

2. 控制台输出触发间隔过长 #

结构化流的控制台Sink默认触发间隔是10秒，如果数据量小或没有新数据，需要等待一段时间才会打印结果。而批处理的show()是立即执行输出。

可以显式设置更短的触发间隔，加快输出速度：

from pyspark.sql.streaming import Trigger

out = df3 \
    .writeStream \
    .format("console") \
    .outputMode("append") \
    .trigger(Trigger.ProcessingTime("1 second"))  # 每秒触发一次
    .start()

3. CSV Schema解析失败导致数据丢失 #

如果df_schema_string定义的Schema和Kafka消息中的CSV格式不匹配，from_csv函数会返回null的stations对象，后续select("stations.*")会得到全null的行，控制台可能不会显示这类无意义的记录。

验证方法：先跳过Schema解析，直接输出原始字符串数据：

df1 = df.selectExpr("CAST(value AS STRING)")
out = df1.writeStream.format("console").outputMode("append").start()
out.awaitTermination()

如果能看到输出，说明问题出在Schema解析环节，需要核对CSV字段顺序、数据类型和df_schema_string是否一致。

4. 多余的上下文对象导致冲突 #

你的代码中同时创建了SparkContext、StreamingContext和SparkSession，这在结构化流场景下完全不必要，甚至可能引发上下文冲突。结构化流只需要SparkSession即可，它会自动管理底层的上下文资源。

修改代码，移除多余的上下文创建代码：

# 移除这两行冗余代码
# sc = SparkContext("local[4]")
# ssc = StreamingContext(sc, 1)

# 直接创建SparkSession并指定运行模式
spark = SparkSession.builder.appName("stations").master("local[4]").getOrCreate()

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