如何将PySpark结构化流DataFrame所有列写入Kafka并解决value列报错

报错原因 #

Spark Structured Streaming 对接Kafka写入时，要求输入的DataFrame必须包含必填的value列（还可按需携带key、partition、topic等可选列），你当前的DataFrame中不存在名为value的字段，所以查询时直接抛出找不到列的异常。

解决方法 #

你需要先将所有列的内容整合为单个value列，最通用的方式是将整行数据序列化为JSON字符串作为value，消费者侧可直接解析JSON还原所有原始字段，修改步骤如下：

1. 导入需要的Spark函数

from pyspark.sql.functions import to_json, struct

2. 调整写入逻辑，同时修正原代码中topic参数缺少引号的问题

dfwriter=df \
  .select(to_json(struct(*df.columns)).alias("value")) \
  .writeStream \
  .format("kafka") \
  .option("checkpointLocation", "/Documents/checkpoint/logs") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("failOnDataLoss", "false") \
  .option("topic", "detection") \
  .start() 

如果需要给Kafka消息指定key（比如基于源主机做分区），可以同时构造key列：

from pyspark.sql.functions import col
dfwriter=df \
  .select(
      col("SourceComputer").alias("key"), # 可按需修改为其他字段作为消息key
      to_json(struct(*df.columns)).alias("value")
  ) \
  .writeStream \
  .format("kafka") \
  .option("checkpointLocation", "/Documents/checkpoint/logs") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("failOnDataLoss", "false") \
  .option("topic", "detection") \
  .start() 

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