使用 Spark 使用对象数组编写 JSON 输出答案

【问题标题】：Writing JSON output with array of objects using Spark使用 Spark 使用对象数组编写 JSON 输出
【发布时间】：2018-06-05 13:17:06
【问题描述】：

我想使用 spark 过程将 json 结构重新格式化为包含对象数组的结构。我的输入文件包含以下行：

{ "keyvals" : [[1,"a"], [2, "b"]] }, 
{ "keyvals" : [[3,"c"], [4, "d"]] }

我希望我的进程输出

{ "keyvals": [{"id": 1, "value": "a"}, {"id": 2, "value": "c"}] },
{ "keyvals": [{"id": 3, "value": "c"}, {"id": 4, "value": "d"}] }

最好的方法是什么？

查看示例输入，您可以在 scala spark-shell 中运行：

var jsonStrings = Seq("""{"keyvals": [[1,"a"], [2, "b"]] }""", """{ "keyvals" : [[3,"c"], [4, "d"]] }""") 
var inputRDD = sc.parallelize(jsonStrings)
var df = spark.sqlContext.read.json(inputRDD)
// reformat goes here ?
df.write.json("myfile.json")

谢谢

【问题讨论】：

你有什么尝试吗？ to_json 也许？请生成minimal reproducible example。
to_json 如何转换 [[1,"a"], [2, "b"]] => [{"id": 1, "value": "a"}, {" id": 2, "value": "c"}] ?需要对数据结构进行转换。

标签： json scala apache-spark apache-spark-sql

【解决方案1】：

如果您检查架构，您会看到以下结构实际上映射到 array<array<string>>

df.printSchema
// root
//  |-- keyvals: array (nullable = true)
//  |    |-- element: array (containsNull = true)
//  |    |    |-- element: string (containsNull = true)

除非元素的数量是固定的，否则你需要一个udf：

import org.apache.spark.sql.functions._   

case class Record(id: Long, value: String)

val parse = udf((xs: Seq[Seq[String]]) => xs.map {
  case Seq(id, value) => Record(id.toLong, value)
})


val result = df.select(parse($"keyvals").alias("keyvals"))

结果可以转换toJSON

result.toJSON.toDF("keyvals").show(false)
// +-------------------------------------------------------+
// |keyvals                                                |
// +-------------------------------------------------------+
// |{"keyvals":[{"id":1,"value":"a"},{"id":2,"value":"b"}]}|
// |{"keyvals":[{"id":3,"value":"c"},{"id":4,"value":"d"}]}|
// +-------------------------------------------------------+

或使用 JSON 编写器 (result.write.json) 编写。

也可以使用强类型Dataset：

df.as[Seq[Seq[String]]].map { xs => xs.map {
  case Seq(id, value) => Record(id.toLong, value)
}}.toDF("keyvals")

【讨论】：