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

Question

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

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

我需要将这个数组类型的列转换为结构类型，其值为{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")

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

Answer 1

如果我理解正确，可以使用分解和透视的组合：

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>