将火花结构化的流数据帧转换为JSON

Question

我正在使用具有结构的火花结构化流读取流：

col1
col2
col3

经过一些转换后，我想以JSON格式将数据帧写入控制台。我正在尝试以下方法：

df.select(to_json($"*"))
      .writeStream
      .outputMode("append")
      .format("console")
      .start()

但我得到Invalid usage of '*' in expression 'structstojson';

是否有一种方法可以将所有行连接到同一列中，以便能够使用to_json？

预期输出是一个带有一列的数据框，每行都有JSON数据：

{"col1":"val11","col2":"val12","col3":"val13"}
{"col1":"val21","col2":"val22","col3":"val23"}

Answer 1

to_json具有以下定义：

def to_json(e: org.apache.spark.sql.Column): org.apache.spark.sql.Column
def to_json(e: org.apache.spark.sql.Column,options: java.util.Map[String,String]): org.apache.spark.sql.Column
def to_json(e: org.apache.spark.sql.Column,options: Map[String,String]): org.apache.spark.sql.Column

这是我们的数据框：

df.show
+----+----+----+
|col1|col2|col3|
+----+----+----+
|   a|   b|   c|
|   d|   e|   f|
+----+----+----+

您需要创建一个struct，然后在其上调用to_json。类似：

df.select(to_json( struct( df.columns.map(col(_)):_*  ) )  as "json").show(false)
+----------------------------------+
|json                              |
+----------------------------------+
|{"col1":"a","col2":"b","col3":"c"}|
|{"col1":"d","col2":"e","col3":"f"}|
+----------------------------------+

Answer 2

这是一种更直接的方法：

df.toJSON.writeStream
    .queryName("aakriti")
    .outputMode("append")
    .format("console")
    .option("numRows", "1000")
    .option("truncate", false)
    .start();