大家好 ！
我正在嘗試將我的結構化流資料幀發送到我的 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、duration、packetCount、processName、協議、時間、時間戳]；第 1 行 pos 5；
如果我發送一個包含value它作業的列的資料框，我會收到關于我的 kafka 主題消費者的資料。
任何想法將我的資料框與所有列一起發送？
謝謝 ！
uj5u.com熱心網友回復：
value正如錯誤所說，您的資料框沒有列。
您需要將所有列“嵌入”一列下value StructType，然后使用類似的函式to_json，而不是CAST( .. AS STRING)
在 Pyspark 中，這類似于struct(to_json(struct($"*")).as("value")在選擇查詢中
類似問題 -將 spark 資料幀的所有列轉換為 json 格式，然后將 json 格式的資料作為列包含在另一個/父資料幀中
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/345641.html
標籤：阿帕奇火花 火花 阿帕奇卡夫卡 火花结构流 spark-kafka-集成