如何在Spark 2 Scala中将Row转换为json

Question

有没有一种简单的方法将给定的Row对象转换为json？

发现这有关将整个Dataframe转换为json输出: Spark Row到JSON

但我只想将一行转换为json.这是我想要做的伪代码.

更确切地说,我在数据帧中读取json作为输入.我正在生成一个主要基于列的新输出,但是有一个json字段用于所有不适合列的信息.

我的问题是什么是编写这个函数最简单的方法:convertRowToJson()

def convertRowToJson(row: Row): String = ???

def transformVenueTry(row: Row): Try[Venue] = {
  Try({
    val name = row.getString(row.fieldIndex("name"))
    val metadataRow = row.getStruct(row.fieldIndex("meta"))
    val score: Double = calcScore(row)
    val combinedRow: Row = metadataRow ++ ("score" -> score)
    val jsonString: String = convertRowToJson(combinedRow)
    Venue(name = name, json = jsonString)
  })
}

Psidom的解决方案:

def convertRowToJSON(row: Row): String = {
    val m = row.getValuesMap(row.schema.fieldNames)
    JSONObject(m).toString()
}

仅当Row只有一个级别而不是嵌套Row时才有效.这是架构:

StructType(
    StructField(indicator,StringType,true),   
    StructField(range,
    StructType(
        StructField(currency_code,StringType,true),
        StructField(maxrate,LongType,true), 
        StructField(minrate,LongType,true)),true))

还尝试了Artem的建议,但是没有编译:

def row2DataFrame(row: Row, sqlContext: SQLContext): DataFrame = {
  val sparkContext = sqlContext.sparkContext
  import sparkContext._
  import sqlContext.implicits._
  import sqlContext._
  val rowRDD: RDD[Row] = sqlContext.sparkContext.makeRDD(row :: Nil)
  val dataFrame = rowRDD.toDF() //XXX does not compile
  dataFrame
}

Answer 1

您可以使用getValuesMap将行对象转换为Map,然后将其转换为JSON:

import scala.util.parsing.json.JSONObject
import org.apache.spark.sql._

val df = Seq((1,2,3),(2,3,4)).toDF("A", "B", "C")    
val row = df.first()          // this is an example row object

def convertRowToJSON(row: Row): String = {
    val m = row.getValuesMap(row.schema.fieldNames)
    JSONObject(m).toString()
}

convertRowToJSON(row)
// res46: String = {"A" : 1, "B" : 2, "C" : 3}

Answer 2

我需要读取 json 输入并生成 json 输出。大多数字段都是单独处理的，但一些 json 子对象只需要保留即可。

当 Spark 读取数据帧时，它会将记录转换为行。Row 是一个类似 json 的结构。可以将其转换并写入 json。

但我需要将一些子 json 结构取出到字符串中以用作新字段。

这可以这样做：

dataFrameWithJsonField = dataFrame.withColumn("address_json", to_json($"location.address"))

location.address是传入的基于 json 的数据帧的子 json 对象的路径。address_json是该对象的列名称，转换为 json 的字符串版本。

to_jsonSpark 2.1 中实现。

如果使用 json4s address_json 生成输出 json，则应将其解析为 AST 表示形式，否则输出 json 将会转义 address_json 部分。