在 Apache Spark 中，如何从爆炸的 JSON 数组中合并多个 SQL 列？答案

【问题标题】：In Apache Spark how could I merge multiple SQL columns from an exploded JSON Array?在 Apache Spark 中，如何从爆炸的 JSON 数组中合并多个 SQL 列？
【发布时间】：2017-08-13 04:58:29
【问题描述】：

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

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|

Approach 的结果如下：

+----+---------+---------+
|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|

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

非常感谢您的帮助！

【问题讨论】：

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

【解决方案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|
+-----+--------+---------+

【讨论】：