Scala Spark - 如何减少在单列中包含多列的数据帧？答案

【问题标题】：Scala Spark - How reduce a dataframe with many couple columns in a single couple columns?Scala Spark - 如何减少在单列中包含多列的数据帧？
【发布时间】：2017-03-30 08:55:27
【问题描述】：

我有一个包含许多（计数和得分）列的数据框。这种情况不是枢轴，而是类似的非枢轴。示例：

|house_score | house_count | mobile_score | mobile_count | sport_score | sport_count | ....<other couple columns>.....| 
|   20            2              48              6             6             78     |
|   40            78             47              74            69             6     |

我想要一个只有两列的新日期框，得分 e 计数。新的数据框减少了仅有的几列中的所有几列。

_________________
| score | count |
|   20  |   2   |
|   40  |   78  |
|   48  |   6   |
|   47  |   74  |
|   6   |   78  |
|   69  |   6   |
|_______________|

什么是最好的解决方案（优雅的代码/性能）？

【问题讨论】：

This 可能包含您问题的解决方案。您的问题可能重复
Pivot Spark Dataframe的可能重复

标签： scala apache-spark spark-dataframe

【解决方案1】：

您可以在列名上使用 foldLeft 来实现这一点（不包括 _ 之后的部分）。这是相当有效的，因为所有密集的操作都是分布式的，并且代码相当干净简洁。

// df from example
val df = sc.parallelize(List((20,2,48,6,6,78), (40,78,47,74,69,6) )).toDF("house_score", "house_count", "mobile_score", "mobile_count", "sport_score", "sport_count")

// grab column names (part before the _)
val cols = df.columns.map(col => col.split("_")(0)).distinct

// fold left over all columns
val result = cols.tail.foldLeft( 
   // init with cols.head column
   df.select(col(s"${cols.head}_score").as("score"), col(s"${cols.head}_count").as("count")) 
){case (acc,c) => {
   // union current column c
   acc.unionAll(df.select(col(s"${c}_score").as("score"),     col(s"${c}_count").as("count")))
}}

result.show

【讨论】：

【解决方案2】：

按照另一个答案中的建议使用 unionAlls 将要求您多次扫描数据，并且在每次扫描项目中，df 仅包含 2 列。从性能的角度来看，如果您可以一次性完成工作，则应避免多次扫描数据，尤其是当您拥有不可缓存的大型数据集或需要进行多次扫描时。

您可以通过生成所有元组（分数、计数）然后对它们进行平面映射，在 1 次中完成。我让你决定它有多优雅：

scala> :paste
// Entering paste mode (ctrl-D to finish)

val df = List((20,2,48,6,6,78), (40,78,47,74,69,6))
    .toDF("house_score", "house_count", "mobile_score", "mobile_count", "sport_score", "sport_count")

df.show

val result = df
    .flatMap(r => Range(0, 5, 2).map(i => (r.getInt(i), r.getInt(i + 1))))
    .toDF("score", "count")

result.show


// Exiting paste mode, now interpreting.

+-----------+-----------+------------+------------+-----------+-----------+
|house_score|house_count|mobile_score|mobile_count|sport_score|sport_count|
+-----------+-----------+------------+------------+-----------+-----------+
|         20|          2|          48|           6|          6|         78|
|         40|         78|          47|          74|         69|          6|
+-----------+-----------+------------+------------+-----------+-----------+

+-----+-----+
|score|count|
+-----+-----+
|   20|    2|
|   48|    6|
|    6|   78|
|   40|   78|
|   47|   74|
|   69|    6|
+-----+-----+

df: org.apache.spark.sql.DataFrame = [house_score: int, house_count: int ... 4 more fields]
result: org.apache.spark.sql.DataFrame = [score: int, count: int]

【讨论】：