java – Spark结构化流：将行转换为json

2023年2月20日 164次阅读

我正在尝试使用spark API将Row of DataFrame转换为json字符串.

从输入行

+----------------+-----------+
|       someThing|       else|
+----------------+-----------+
|            life|         42|
+----------------+-----------+

同

myDataFrame
.select(struct("*").as("col"))
.select(to_json(col("col")))
.writeStream()
.foreach(new KafkaWriter())
.start()

使用KafkaWriter,即使用row.toString()我得到：

[{  
    "someThing":"life",
    "else":42
}]

当我想要得到这个：

{  
    "someThing":"life",
    "else":42
}

(没有[])

任何的想法？

最佳答案 刚刚找到解决方案.使用Row.mkString而不是Row.toString解决了我的问题.

点赞