如何从结构类型的数组创建结构？答案

【问题标题】：How to create a struct from an array of struct type?如何从结构类型的数组创建结构？
【发布时间】：2019-12-03 03:07:15
【问题描述】：

如何合并数组中的所有结构并生成合并的结构？

例如，使用 Spark SQL，我能够读取源 json 文件并生成数组类型的列，数组中的每个元素都包含一个键：值对。例如，假设我们有值列名为 col 和值：[{a: 1}, {b: 2}, {c: 3}]

的 ArrayType 列

我需要将此数组类型列转换为结构类型，值为{a: 1, b: 2, c: 3}。

由于我是通过读取 JSON 文件获得架构，因此我可以使用派生序数获得结果，例如：

df.select(
  $"col.a".getItem(1) as "a", 
  $"col.b".getItem(2) as "b", 
  $"col.c".getItem(3) as "c")

这个解决方案的问题是，如果数组中的元素排序不同，我会得到错误的结果。有没有一种干净的方法来合并所有键/值结构并干净地生成一个结构？就我而言，键没有重复项，因此我不必担心在键/值被覆盖时会丢失数据。

【问题讨论】：

标签： apache-spark-sql

【解决方案1】：

如果我理解正确，可以使用explode 和pivot 的组合来完成：

scala> :paste
// Entering paste mode (ctrl-D to finish)

val df = Seq(
  (1, Array(("a", 1), ("b", 2), ("c", 3))),
  (2, Array(("b", 5), ("c", 6), ("a", 4)))
).toDF("id", "col")

df.show(10, false)

val explodedDF = df.withColumn("col2", explode(df.col("col"))).select("id", "col2")

explodedDF.show(10, false)

val flattenedDF = explodedDF.withColumn("k", $"col2._1").withColumn("v", $"col2._2").select("id", "k", "v")

flattenedDF.show(10, false)

val pivotedDF = flattenedDF.groupBy("id").pivot("k").agg(first(col("v")))

pivotedDF.show(10, false)

import scala.util.parsing.json.JSONObject

pivotedDF.select("a", "b", "c").collect().map{row => JSONObject(row.getValuesMap(row.schema.fieldNames))}.map(println)

// Exiting paste mode, now interpreting.

+---+------------------------+
|id |col                     |
+---+------------------------+
|1  |[[a, 1], [b, 2], [c, 3]]|
|2  |[[b, 5], [c, 6], [a, 4]]|
+---+------------------------+

+---+------+
|id |col2  |
+---+------+
|1  |[a, 1]|
|1  |[b, 2]|
|1  |[c, 3]|
|2  |[b, 5]|
|2  |[c, 6]|
|2  |[a, 4]|
+---+------+

+---+---+---+
|id |k  |v  |
+---+---+---+
|1  |a  |1  |
|1  |b  |2  |
|1  |c  |3  |
|2  |b  |5  |
|2  |c  |6  |
|2  |a  |4  |
+---+---+---+

+---+---+---+---+
|id |a  |b  |c  |
+---+---+---+---+
|1  |1  |2  |3  |
|2  |4  |5  |6  |
+---+---+---+---+

{"a" : 1, "b" : 2, "c" : 3}
{"a" : 4, "b" : 5, "c" : 6}
df: org.apache.spark.sql.DataFrame = [id: int, col: array<struct<_1:string,_2:int>>]
explodedDF: org.apache.spark.sql.DataFrame = [id: int, col2: struct<_1: string, _2: int>]
flattenedDF: org.apache.spark.sql.DataFrame = [id: int, k: string ... 1 more field]
pivotedDF: org.apache.spark.sql.DataFrame = [id: int, a: int ... 2 more fields]
import scala.util.parsing.json.JSONObject
res24: Array[Unit] = Array((), ())

scala>

【讨论】：