PySpark读取Kafka流触发Queries with streaming sources错误如何处理

错误根因 #

• 你在Structured Streaming产生的流DataFrame上直接调用了.rdd方法转为RDD操作，流DataFrame不支持这类即时触发的算子，所有流处理逻辑必须通过流写入启动（writeStream.start()）才会执行，调用.rdd会触发立即执行计算，直接抛出该异常。
• 附加代码问题：你已经通过selectExpr("CAST(value AS STRING)")得到了字符串类型的rawDF，但后续拆分逻辑仍然使用了原始二进制类型的inputDF.value，就算不报流查询错误也会出现类型不匹配问题；另外RDD的方法是小写.map()不是.Map()，写法错误。

修复方案 #

不要使用RDD转换，直接使用Structured Streaming内置的字符串处理、类型转换算子实现字段解析，修复后代码如下：

from dataclasses import dataclass
from pyspark.sql import SparkSession
import pyspark.sql.functions as f
from pyspark.sql.types import StringType, FloatType, StructType, StructField

@dataclass
class DeviceData:
    device: str
    temp: float
    humd: float
    pres: float
    
# 提前定义解析后的Schema，避免RDD转换
device_schema = StructType([
    StructField("device", StringType(), nullable=False),
    StructField("temp", FloatType(), nullable=False),
    StructField("humd", FloatType(), nullable=False),
    StructField("pres", FloatType(), nullable=False)
])

spark:SparkSession = SparkSession.builder \
    .master("local[1]") \
    .appName("StreamHandler") \
    .getOrCreate()

spark.sparkContext.setLogLevel("WARN")

inputDF = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "weather") \
    .load()

# 先将value转成字符串，再拆分、类型转换，全程用DataFrame算子
rawDF = inputDF.selectExpr("CAST(value AS STRING) as value_str")
df_split = rawDF.select(
    f.split(f.col("value_str"), ",").alias("fields")
).select(
    f.col("fields")[0].cast(StringType()).alias("device"),
    f.col("fields")[1].cast(FloatType()).alias("temp"),
    f.col("fields")[2].cast(FloatType()).alias("humd"),
    f.col("fields")[3].cast(FloatType()).alias("pres")
)

summaryDF = df_split.groupBy('device') \
    .agg(f.avg('temp'), f.avg('humd'), f.avg('pres'))

query = summaryDF.writeStream.format('console').outputMode('update').start()
query.awaitTermination()

关键改动说明 #

• 全程使用DataFrame内置算子处理流数据，完全避免RDD转换，符合Structured Streaming的执行要求
• 直接通过拆分后的数组下标取对应字段，显式指定字段类型，不需要额外的类映射转换
• 复用了转换为字符串类型的value字段，避免类型不匹配错误

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