热释光；dr使用 struct 函数后跟 to_json （作为 toJSON 由于spark-17029（20天前刚刚修复），流式数据集已损坏）。
引用struct的scaladoc:
struct（colname:string，colnames:string*）：column创建一个由多个输入列组成的新struct列。
如果您使用java api，那么struct function也有4种不同的变体：
公共静态列结构（列。。。cols）创建新的结构列。
使用tojson函数，您的案例包括：
public static column to \u json（column e）将包含structtype的列转换为具有指定模式的json字符串。
以下是scala代码（将其转换为java是您的家庭练习）：

val ds1 = Seq(("hello", 1.56, 34)).toDF("name", "ratio", "count")
val recordCol = to_json(struct("name", "ratio", "count")) as "record"
scala> ds1.select(recordCol).show(truncate = false)
+----------------------------------------+
|record                                  |
+----------------------------------------+
|{"name":"hello","ratio":1.56,"count":34}|
+----------------------------------------+

我还尝试了一下您的解决方案（使用今天构建的spark2.3.0-snapshot），它似乎工作得非常完美。

val fromKafka = spark.
  readStream.
  format("kafka").
  option("subscribe", "topic1").
  option("kafka.bootstrap.servers", "localhost:9092").
  load.
  select('value cast "string")
fromKafka.
  toJSON. // <-- JSON conversion
  writeStream.
  format("console"). // using console sink
  start
``` `format("kafka")` 在spark-19719中添加，在2.1.0中不可用。