【问题标题】：spark normalize data frame of arraysspark规范化数组的数据帧
【发布时间】：2018-03-31 08:01:02
【问题描述】：

如何规范化主要由嵌套数组组成的 spark 数据帧？

case class FooBar(id:String, foo:Seq[String], bar:String, baz: Seq[String])
val f = Seq(FooBar("thinga", Seq("1 "), "1 2 3 ", Seq("2 ")),
    FooBar("thinga", Seq("1 2 3 4 "), " 0 0 0 ", Seq("2 3 4 5 ")),
    FooBar("thingb", Seq("1 2 "), "1 2 3 4 5 ", Seq("1 2 ")),
    FooBar("thingb", Seq("0 ", "0 ", "0 "), "1 2 3 4 5 ", Seq("1 2 3 "))).toDS
f.printSchema
f.show(false)
+------+------------+----------+----------+
|    id|         foo|       bar|       baz|
+------+------------+----------+----------+
|thinga|        [1 ]|    1 2 3 |      [2 ]|
|thinga|  [1 2 3 4 ]|    0 0 0 |[2 3 4 5 ]|
|thingb|      [1 2 ]|1 2 3 4 5 |    [1 2 ]|
|thingb|[0 , 0 , 0 ]|1 2 3 4 5 |  [1 2 3 ]|
+------+------------+----------+----------+


scala> f.printSchema
root
 |-- id: string (nullable = true)
 |-- foo: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- bar: string (nullable = true)
 |-- baz: array (nullable = true)
 |    |-- element: string (containsNull = true)

我想要一个像explode这样的东西，它将保留（id，foo，bar，baz）的模式，但为数组的每个值返回一个单独的记录。最终结果不应再包含数组。

Foo 和 baz 是相关的。他们的顺序不能被扭曲。它们总是具有相同的长度，并且 foo 的第一个值与 baz 的第一个值相关 - 依此类推。也许我应该先将它们组合成一个列/结构？

最终结果应该类似于：

+------+------------+----------+----------+
|    id|         foo|       bar|       baz|
+------+------------+----------+----------+
|thinga|        1   |    1     |      2   |
|thinga|        1   |      2   |      2   |
|thinga|        1   |        3 |      2   |    
|thinga|  1         |    0     |2         |
|thinga|  2         |    0     |3         |
|thinga|  3         |    0     |4         |
|thinga|  4         |    0     |5         |
|thinga|  1         |    0     |2         |
|thinga|  2         |    0     |3         |
|thinga|  3         |    0     |4         |
|thinga|  4         |    0     |5         |

|thinga|  1         |    0     |2         |
|thinga|  2         |    0     |3         |
|thinga|  3         |    0     |4         |
|thinga|  4         |    0     |5         |

....

|thingb|0           |1         |       1  |
|thingb|0           |2         |       2  |
|thingb|0           |3         |       3  |
|thingb|0           |4         |       1  |
|thingb|0           |5         |       2  |
|thingb|0           |1         |       3  |
|thingb|0           |2         |       1  |
|thingb|0           |3         |       2  |
|thingb|0           |4         |       3  |
|thingb|0           |5         |       1  |
|thingb|0           |1         |       2  |
|thingb|0           |2         |       3  |
|thingb|0           |3         |       1  |
|thingb|0           |4         |       2  |
|thingb|0           |5         |       3  |
+------+------------+----------+----------+

编辑

【问题讨论】：

你看过posexplode()spark.apache.org/docs/latest/api/scala/…
我玩过它 - 但还没有达到预期的效果。
您的数据很奇怪，您的数组仅包含 1 个元素（空格分隔），bar 在您的示例中也不是数组，但您仍想“爆炸”它？
你好@GeorgHeiler 如果你的数据被美化成这样，那将是一个问题：val f = Seq(FooBar("thinga", Seq("1"), "1 2 3", Seq("2")), FooBar("thinga", Seq("1", "2", "3", "4"), "0 0 0", Seq("2", "3", "4", "5")), FooBar("thingb", Seq("1", "2"), "1 2 3 4 5", Seq("1", "2")), FooBar("thingb", Seq("0", "0", "0"), "1 2 3 4 5", Seq("1", "2", "3"))).toDS()
这将从 seq 中删除空间，并将在 seq 中包含多个项目而不是一个项目

标签： arrays apache-spark struct apache-spark-sql spark-dataframe

【解决方案1】：

根据我们的讨论（请检查初始帖子下的 cmets），以下数据应该是有效的：

+------+-------+---------+-------+
|    id|    foo|      bar|    baz|
+------+-------+---------+-------+
|thinga|      1|    1 2 3|      2|
|thinga|1 2 3 4|    0 0 0|2 3 4 5|
|thingb|    1 2|1 2 3 4 5|    1 2|
|thingb|  0 0 0|1 2 3 4 5|  1 2 3|
+------+-------+---------+-------+

然后初始化代码如下：

case class FooBar(id:String, foo:String, bar:String, baz: String)
val f = Seq(FooBar("thinga", "1", "1 2 3", "2"),
  FooBar("thinga", "1 2 3 4", "0 0 0", "2 3 4 5"),
  FooBar("thingb", "1 2", "1 2 3 4 5", "1 2"),
  FooBar("thingb", "0 0 0", "1 2 3 4 5", "1 2 3")).toDS()

如果是这种情况，那么这段代码应该会产生理想的结果：

f
  .withColumn("foo", split($"foo", " "))
  .withColumn("baz", split($"baz", " "))
  .withColumn("bar", explode(split($"bar", " ")))
  .map { case Row(id: String, foo: Seq[String], bar: String, baz: Seq[String]) =>
    val c = for ((f, b) <- foo.zip(baz)) yield {
      (f, b)
    }
    (id, bar, c)
  }.toDF(cols: _*)
  .withColumn("foo+baz", explode($"foo+baz"))
  .withColumn("foo", $"foo+baz._1")
  .withColumn("baz", $"foo+baz._2")
  .drop($"foo+bar")
  .select("id", "foo", "bar", "baz")
  .show(100)

前两个转换将拆分空格分隔的列 foo 和 baz。由于列 bar 是字符串，我们需要将其转换为具有拆分的数组，然后将其分解。 Map 将返回 (id, bar, c) 的元组，其中 c 是元组 (foo, bar) 的序列。映射后我们得到下一个输出：

+------+---+--------------------+
|    id|bar|             foo+baz|
+------+---+--------------------+
|thinga|  1|             [[1,2]]|
|thinga|  2|             [[1,2]]|
|thinga|  3|             [[1,2]]|
|thinga|  0|[[1,2], [2,3], [3...|
|thinga|  0|[[1,2], [2,3], [3...|
|thinga|  0|[[1,2], [2,3], [3...|
|thingb|  1|      [[1,1], [2,2]]|
|thingb|  2|      [[1,1], [2,2]]|
|thingb|  3|      [[1,1], [2,2]]|
|thingb|  4|      [[1,1], [2,2]]|
|thingb|  5|      [[1,1], [2,2]]|
|thingb|  1|[[0,1], [0,2], [0...|
|thingb|  2|[[0,1], [0,2], [0...|
|thingb|  3|[[0,1], [0,2], [0...|
|thingb|  4|[[0,1], [0,2], [0...|
|thingb|  5|[[0,1], [0,2], [0...|
+------+---+--------------------+

接下来我们用 "foo+baz" 再做一次分解以提取最终的元组。现在输出如下：

+------+---+-------+
|    id|bar|foo+baz|
+------+---+-------+
|thinga|  1|  [1,2]|
|thinga|  2|  [1,2]|
|thinga|  3|  [1,2]|
|thinga|  0|  [1,2]|
|thinga|  0|  [2,3]|
|thinga|  0|  [3,4]|
|thinga|  0|  [4,5]|
|thinga|  0|  [1,2]|
.....
|thingb|  1|  [0,2]|
|thingb|  1|  [0,3]|
|thingb|  2|  [0,1]|
|thingb|  2|  [0,2]|
|thingb|  2|  [0,3]|
|thingb|  3|  [0,1]|
|thingb|  3|  [0,2]|
|thingb|  3|  [0,3]|
|thingb|  4|  [0,1]|
|thingb|  4|  [0,2]|
|thingb|  4|  [0,3]|
|thingb|  5|  [0,1]|
|thingb|  5|  [0,2]|
|thingb|  5|  [0,3]|
+------+---+-------+

最后，我们从 foo+baz._1 和 foo+baz._2 中填充 foo 和 baz 列> 分别。这将是最终输出：

+------+---+---+---+
|    id|foo|bar|baz|
+------+---+---+---+
|thinga|  1|  1|  2|
|thinga|  1|  2|  2|
|thinga|  1|  3|  2|
|thinga|  1|  0|  2|
|thinga|  2|  0|  3|
|thinga|  3|  0|  4|
|thinga|  4|  0|  5|
|thinga|  1|  0|  2|
|thinga|  2|  0|  3|
|thinga|  3|  0|  4|
|thinga|  4|  0|  5|
|thinga|  1|  0|  2|
|thinga|  2|  0|  3|
|thinga|  3|  0|  4|
|thinga|  4|  0|  5|
|thingb|  1|  1|  1|
|thingb|  2|  1|  2|
|thingb|  1|  2|  1|
|thingb|  2|  2|  2|
|thingb|  1|  3|  1|
|thingb|  2|  3|  2|
|thingb|  1|  4|  1|
|thingb|  2|  4|  2|
|thingb|  1|  5|  1|
|thingb|  2|  5|  2|
|thingb|  0|  1|  1|
|thingb|  0|  1|  2|
|thingb|  0|  1|  3|
|thingb|  0|  2|  1|
|thingb|  0|  2|  2|
|thingb|  0|  2|  3|
|thingb|  0|  3|  1|
|thingb|  0|  3|  2|
|thingb|  0|  3|  3|
|thingb|  0|  4|  1|
|thingb|  0|  4|  2|
|thingb|  0|  4|  3|
|thingb|  0|  5|  1|
|thingb|  0|  5|  2|
|thingb|  0|  5|  3|
+------+---+---+---+

【讨论】：