spark规范化数组的数据帧
Posted
技术标签:
【中文标题】spark规范化数组的数据帧【英文标题】：spark normalize data frame of arrays 【发布时间】：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 |
+------+------------+----------+----------+
编辑
部分相关的问题 - Explode (transpose?) multiple columns in Spark SQL table
【问题讨论】：
你看过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 中包含多个项目而不是一个项目
【参考方案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|
+------+---+---+---+
【讨论】：
以上是关于spark规范化数组的数据帧的主要内容，如果未能解决你的问题，请参考以下文章
使用 StandardScaler() 规范化 pandas 数据帧，不包括特定列
如何在 pandas 数据帧中有效地使用 one-hot 编码规范化列？
使用 reduceByKey(numPartitions) 或 repartition 规范化 SPARK RDD 分区