如何分解列

Question

之后：

val df = Seq((1, Vector(2, 3, 4)), (1, Vector(2, 3, 4))).toDF("Col1", "Col2")

我在Apache Spark中有这个数据帧：

+------+---------+
| Col1 | Col2    |
+------+---------+
|  1   |[2, 3, 4]|
|  1   |[2, 3, 4]|
+------+---------+

如何将其转换为：

+------+------+------+------+
| Col1 | Col2 | Col3 | Col4 |
+------+------+------+------+
|  1   |  2   |  3   |  4   |
|  1   |  2   |  3   |  4   |
+------+------+------+------+

Answer 1

一个不与RDD相互转换的解决方案：

df.select($"Col1", $"Col2"(0) as "Col2", $"Col2"(1) as "Col3", $"Col2"(2) as "Col3")

或者有争议更好：

val nElements = 3
df.select(($"Col1" +: Range(0, nElements).map(idx => $"Col2"(idx) as "Col" + (idx + 2)):_*))

Spark 数组列的大小不是固定的，例如，您可以有：

+----+------------+
|Col1|        Col2|
+----+------------+
|   1|   [2, 3, 4]|
|   1|[2, 3, 4, 5]|
+----+------------+

因此，无法获取列的数量并创建它们。如果您知道大小始终相同，则可以像这样设置nElements：

val nElements = df.select("Col2").first.getList(0).size

Answer 2

只是为了给出 sgvd 的 Pyspark 版本的答案。如果数组列在 Col2 中，则此 select 语句会将每个数组的前nElements Col2移动到它们自己的列中：

from pyspark.sql import functions as F            
df.select([F.col('Col2').getItem(i) for i in range(nElements)])

Answer 3

只需添加到 sgvd 的解决方案中：

如果大小并不总是相同，则可以像这样设置 nElements：

val nElements = df.select(size('Col2).as("Col2_count"))
                  .select(max("Col2_count"))
                  .first.getInt(0)

Answer 4

您可以使用地图：

df.map {
    case Row(col1: Int, col2: mutable.WrappedArray[Int]) => (col1, col2(0), col2(1), col2(2))
}.toDF("Col1", "Col2", "Col3", "Col4").show()

Answer 5

如果您正在使用SparkR，您可以在此处找到我的答案，您不需要使用 explode 但您需要SparkR::dapply和stringr::str_split_fixed。