将spark结构化流数据帧转换为JSON

Posted 2021-04-18

我正在使用具有以下结构的spark结构化流读取流：

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"}

答案

qazxsw poi有以下定义：

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