Spark 在 Scala 中爆炸 - 将爆炸列添加到行

Posted 2023-03-23

【中文标题】Spark 在 Scala 中爆炸 - 将爆炸列添加到行

【英文标题】：Spark explode in Scala - Add exploded column to the row

【发布时间】：2022-01-16 16:56:42

【问题描述】：

我有一个包含以下内容的 Spark Dataframe：

Name	E1	E2	E3
abc	4	5	6

我需要各种E 列成为新列中的行，如下所示：

Name	value	EType
abc	4	E1
abc	5	E2
abc	6	E3

This answer 给了我使用explode 的想法，我现在有了以下代码：

df.select($"Name", explode(array("E1", "E2", "E3")).as("value"))

上面的代码为我提供了我需要的 Name 和 value 列，但我仍然需要一种方法来添加 EType 列，基于传递给 explode 的数组中的哪个值被用于填充该特定行。

以上代码的输出：

Name	value
abc	4
abc	5
abc	6

如何添加 Etype 列？

（我正在使用带有 Scala 的 Spark 2.2）

谢谢！

【问题讨论】：

这能回答你的问题吗？ How to melt Spark DataFrame?

【参考方案1】：

对于这种特殊情况，您可以使用stack 函数。

df.selectExpr('Name', "stack(3, E1, 'E1', E2, 'E2', E3, 'E3')").toDF('Name', 'value', 'EType').show()

df.selectExpr('Name', "stack(3, E1, 'E1', E2, 'E2', E3, 'E3')").toDF('Name', 'value', 'EType').show()
df.selectExpr('Name', "stack(3, E1, 'E1', E2, 'E2', E3, 'E3')").toDF('Name', 'value', 'EType').show()
+----+-----+-----+
|Name|value|EType|
+----+-----+-----+
| abc|    4|   E1|
| abc|    5|   E2|
| abc|    6|   E3|
+----+-----+-----+

【参考方案2】：

您可以分解包含列名称及其内容的struct，而不是仅仅分解值，如下所示：

import org.apache.spark.sql.functions.array, col, explode, lit, struct

val result = df
  .select(
    col("name"), 
    explode(array(
      df.columns.filterNot(_ == "name").map(c => struct(lit(c).as("EType"), col(c).alias("value"))): _*
    ))
  )
  .select("name", "col.*")

根据您的输入，您将获得 result 数据框：

+----+-----+-----+
|name|EType|value|
+----+-----+-----+
|abc |E1   |4    |
|abc |E2   |5    |
|abc |E3   |6    |
+----+-----+-----+

【参考方案3】：

这里需要用到melt操作。

注意：pyspark 中不存在 Melt 功能，您需要编写该 util 函数。

你可以去想这个关于如何实现融化功能的答案How to melt Spark DataFrame?