如何将我的结构化流数据帧发送到 kafka？

Question

大家好！

我正在尝试将我的结构化流数据帧发送到我的 kafka 主题之一，detection。

这是结构化流数据帧的架构：

 root
 |-- timestamp: timestamp (nullable = true)
 |-- Sigma: string (nullable = true)
 |-- time: string (nullable = true)
 |-- duration: string (nullable = true)
 |-- SourceComputer: string (nullable = true)
 |-- SourcePort: string (nullable = true)
 |-- DestinationComputer: string (nullable = true)
 |-- DestinationPort: string (nullable = false)
 |-- protocol: string (nullable = true)
 |-- packetCount: string (nullable = true)
 |-- byteCount: string (nullable = true)

然后我尝试使用这种方法发送数据帧：

dfwriter=df \
  .selectExpr("CAST(value AS STRING)") \
  .writeStream \
  .format("kafka") \
  .option("checkpointLocation", "/Documents/checkpoint/logs") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("failOnDataLoss", "false") \
  .option("topic", detection) \
  .start() 

然后我得到了错误：

pyspark.sql.utils.AnalysisException：无法解析“value”给定的输入列：[DestinationComputer、DestinationPort、Sigma、SourceComputer、SourcePort、byteCount、持续时间、packetCount、processName、协议、时间、时间戳] ;第 1 行第 5 行；

如果我发送一个带有 value 列的数据框，它可以工作，我会收到关于我的 kafka 主题消费者的数据。

任何想法发送我的所有列的数据框？

谢谢！

Answer 1

如错误所示，您的数据框没有 value 列。

您需要在value StructType 列下“嵌入”所有列，然后使用to_json 之类的函数，而不是CAST( .. AS STRING)

在 Pyspark 中，这类似于选择查询中的 struct(to_json(struct($"*")).as("value")

类似问题 - Convert all the columns of a spark dataframe into a json format and then include the json formatted data as a column in another/parent dataframe