使用scala在Spark中转置DataFrame而不进行聚合答案

【问题标题】：Transpose DataFrame Without Aggregation in Spark with scala使用scala在Spark中转置DataFrame而不进行聚合
【发布时间】：2018-03-20 19:30:22
【问题描述】：

我在网上查看了许多不同的解决方案，但数数没有找到我想要解决的问题。请帮我解决这个问题。

我正在使用带有 Scala 的 Apache Spark 2.1.0。以下是我的数据框：


+-----------+-------+
|COLUMN_NAME| VALUE |
+-----------+-------+
|col1       | val1  |
|col2       | val2  |
|col3       | val3  |
|col4       | val4  |
|col5       | val5  |
+-----------+-------+

我希望将其转置为，如下所示：


+-----+-------+-----+------+-----+
|col1 | col2  |col3 | col4 |col5 |
+-----+-------+-----+------+-----+
|val1 | val2  |val3 | val4 |val5 |
+-----+-------+-----+------+-----+

【问题讨论】：

如果两条记录具有相同的COLUMN_NAME 但不同的VALUE 怎么办？那应该是什么价值呢？如果你知道没有这样的重复，你的数据框要么非常小（在这种情况下，你可以收集它并使用普通的 Scala 进行转换）或者结果会有太多的列。
两条记录永远不会有相同的列名。事实上，我在多行中获得了表插入/更新详细信息，一列带有 columnn-names，另一列带有值，我的计划是将它们转换为数据框并将它们直接更新到 Kudu 数据库中。第一列值作为模式出现，并发送作为值的列值。所以我需要用它来构建数据框。如果您有任何其他建议/想法，请告诉我。

标签： scala apache-spark dataframe transpose

【解决方案1】：

您可以使用pivot 执行此操作，但您仍然需要聚合，但是如果您有多个value 用于COLUMN_NAME 怎么办？

val df = Seq(
  ("col1", "val1"),
  ("col2", "val2"),
  ("col3", "val3"),
  ("col4", "val4"),
  ("col5", "val5")
).toDF("COLUMN_NAME", "VALUE")

df
  .groupBy()
  .pivot("COLUMN_NAME").agg(first("VALUE"))
  .show()

+----+----+----+----+----+
|col1|col2|col3|col4|col5|
+----+----+----+----+----+
|val1|val2|val3|val4|val5|
+----+----+----+----+----+

编辑：

如果您的数据框真的像您的示例中那样小，您可以将其收集为Map：

val map = df.as[(String,String)].collect().toMap

然后申请this answer

【讨论】：

非常感谢您的快速回复！非常感激！有效：）。唯一的问题是因为枢轴，它很慢。
嘿 Raphael，我知道我们也可以使用 Map 来做到这一点，但我无法获得结果。如果您对 Map 有逻辑，请分享。
@MarutiK 只需在您的地图上拨打toSeq，然后应用我的答案
我可以做 toSeq ，之后它会因 groupBy() 为空值而失败。我收到错误为“ :46: error: not enough arguments for method groupBy: (f: ((String, String)) => K)scala.collection.immutable.Map[K,Seq[(String,细绳）]] ”。对此有任何建议。
@RaphaelRoth 感谢您的回答。如果只有一列怎么办？我为此尝试了这个解决方案，但我得到了一个额外的行，我该如何摆脱它？

【解决方案2】：

如果您的数据框像问题中一样小，那么您可以收集 COLUMN_NAME 以形成架构并收集 VALUE 以形成行然后创建一个新的数据框作为

import org.apache.spark.sql.functions._
import org.apache.spark.sql.Row
//creating schema from existing dataframe
val schema = StructType(df.select(collect_list("COLUMN_NAME")).first().getAs[Seq[String]](0).map(x => StructField(x, StringType)))
//creating RDD[Row] 
val values = sc.parallelize(Seq(Row.fromSeq(df.select(collect_list("VALUE")).first().getAs[Seq[String]](0))))
//new dataframe creation
sqlContext.createDataFrame(values, schema).show(false)

这应该给你

+----+----+----+----+----+
|col1|col2|col3|col4|col5|
+----+----+----+----+----+
|val1|val2|val3|val4|val5|
+----+----+----+----+----+

【讨论】：

你太棒了拉梅什！这正是我所需要的。非常感谢你的帮助。在性能方面，这比 Pivot 更好。
很高兴听到@MarutiK，当你有资格时不要忘记投票;)

【解决方案3】：

使用交叉表的另一种解决方案虽然冗长。

 val dfp = spark.sql(""" with t1 (
 select  'col1' c1, 'val1' c2  union all
 select  'col2' c1, 'val2' c2  union all
 select  'col3' c1, 'val3' c2  union all
 select  'col4' c1, 'val4' c2  union all
 select  'col5' c1, 'val5' c2
  )  select   c1  COLUMN_NAME,   c2  VALUE     from t1
""")
dfp.show(50,false)

+-----------+-----+
|COLUMN_NAME|VALUE|
+-----------+-----+
|col1       |val1 |
|col2       |val2 |
|col3       |val3 |
|col4       |val4 |
|col5       |val5 |
+-----------+-----+

val dfp2=dfp.groupBy("column_name").agg( first($"value") as "value" ).stat.crosstab("value", "column_name")
dfp2.show(false)

+-----------------+----+----+----+----+----+
|value_column_name|col1|col2|col3|col4|col5|
+-----------------+----+----+----+----+----+
|val1             |1   |0   |0   |0   |0   |
|val3             |0   |0   |1   |0   |0   |
|val2             |0   |1   |0   |0   |0   |
|val5             |0   |0   |0   |0   |1   |
|val4             |0   |0   |0   |1   |0   |
+-----------------+----+----+----+----+----+

val needed_cols = dfp2.columns.drop(1)

needed_cols: Array[String] = Array(col1, col2, col3, col4, col5)

val dfp3 = needed_cols.foldLeft(dfp2) { (acc,x) => acc.withColumn(x,expr(s"case when ${x}=1 then value_column_name else 0 end")) }
dfp3.show(false)

+-----------------+----+----+----+----+----+
|value_column_name|col1|col2|col3|col4|col5|
+-----------------+----+----+----+----+----+
|val1             |val1|0   |0   |0   |0   |
|val3             |0   |0   |val3|0   |0   |
|val2             |0   |val2|0   |0   |0   |
|val5             |0   |0   |0   |0   |val5|
|val4             |0   |0   |0   |val4|0   |
+-----------------+----+----+----+----+----+

dfp3.select( needed_cols.map( c => max(col(c)).as(c)) :_* ).show

+----+----+----+----+----+
|col1|col2|col3|col4|col5|
+----+----+----+----+----+
|val1|val2|val3|val4|val5|
+----+----+----+----+----+

【讨论】：

【解决方案4】：

要增强 Ramesh Maharjan 的答案，请收集并将其转换为地图。

val mp = df.as[(String,String)].collect.toMap

使用虚拟数据框，我们可以使用 foldLeft 进一步构建

val f = Seq("1").toDF("dummy")

mp.keys.toList.sorted.foldLeft(f) { (acc,x) => acc.withColumn(mp(x),lit(x) ) }.drop("dummy").show(false)

+----+----+----+----+----+
|val1|val2|val3|val4|val5|
+----+----+----+----+----+
|col1|col2|col3|col4|col5|
+----+----+----+----+----+

【讨论】：