在 Apache Spark 中，如何合并松散 JSON 数组中的多个 SQL 列?

Question

>我正在从一个目录中读取多个JSON文件;这个JSON在一个数组中有多个项目'cars'。我正在尝试分解并将项目"汽车"中的离散值合并到一个数据帧。

JSON 文件如下所示：

{
"cars": {
"items": 
[
{
"latitude": 42.0001,
"longitude": 19.0001,
"name": "Alex"
},
{
"latitude": 42.0002,
"longitude": 19.0002,
"name": "Berta"
},
{
"latitude": 42.0003,
"longitude": 19.0003,
"name": "Chris"
},
{
"latitude": 42.0004,
"longitude": 19.0004,
"name": "Diana"
}
]
}
}

我将值分解并合并到一个数据帧的方法是：

// Read JSON files
val jsonData = sqlContext.read.json(s"/mnt/$MountName/.")
// To sqlContext to DataFrame
val jsonDF = jsonData.toDF()
/* Approach 1 */
// User-defined function to 'zip' two columns
val zip = udf((xs: Seq[Double], ys: Seq[Double]) => xs.zip(ys))
jsonDF.withColumn("vars", explode(zip($"cars.items.latitude", $"cars.items.longitude"))).select($"cars.items.name", $"vars._1".alias("varA"), $"vars._2".alias("varB"))
/* Apporach 2 */
val df = jsonData.select($"cars.items.name", $"cars.items.latitude", $"cars.items.longitude").toDF("name", "latitude", "longitude")
val df1 = df.select(explode(df("name")).alias("name"), df("latitude"), df("longitude"))
val df2 = df1.select(df1("name").alias("name"), explode(df1("latitude")).alias("latitude"), df1("longitude"))
val df3 = df2.select(df2("name"), df2("latitude"), explode(df2("longitude")).alias("longitude"))

如您所见，方法 1的结果只是两个离散"合并"参数的数据帧，例如：

+--------------------+---------+---------+
|                name|     varA|     varB|
+--------------------+---------+---------+
|[Leo, Britta, Gor...|48.161079|11.556778|
|[Leo, Britta, Gor...|48.124666|11.617682|
|[Leo, Britta, Gor...|48.352043|11.788091|
|[Leo, Britta, Gor...| 48.25184|11.636337|

方法的结果如下：

+----+---------+---------+
|name| latitude|longitude|
+----+---------+---------+
| Leo|48.161079|11.556778|
| Leo|48.161079|11.617682|
| Leo|48.161079|11.788091|
| Leo|48.161079|11.636337|
| Leo|48.161079|11.560595|
| Leo|48.161079|11.788632|

(结果是每个"名称"与每个"纬度"与每个"经度"的映射)

结果应如下所示：

+--------------------+---------+---------+
|                name|     varA|     varB|
+--------------------+---------+---------+
|Leo                 |48.161079|11.556778|
|Britta              |48.124666|11.617682|
|Gorch               |48.352043|11.788091|

您知道如何读取文件，拆分和合并每行只是一个对象的值吗？

非常感谢您的帮助！

Answer 1

为了获得预期的结果，您可以尝试以下方法：

// Read JSON files
val jsonData = sqlContext.read.json(s"/mnt/$MountName/.")
// To sqlContext to DataFrame
val jsonDF = jsonData.toDF()
// Approach
val df1 = jsonDF.select(explode(df("cars.items")).alias("items"))
val df2 = df1.select("items.name", "items.latitude", "items.longitude")

上述方法将为您提供以下结果：

+-----+--------+---------+
| name|latitude|longitude|
+-----+--------+---------+
| Alex| 42.0001|  19.0001|
|Berta| 42.0002|  19.0002|
|Chris| 42.0003|  19.0003|
|Diana| 42.0004|  19.0004|
+-----+--------+---------+