如何在 Spark 1.5 中转置数据帧（没有可用的枢轴运算符）？答案

【问题标题】：How to transpose dataframe in Spark 1.5 (no pivot operator available)?如何在 Spark 1.5 中转置数据帧（没有可用的枢轴运算符）？
【发布时间】：2016-07-12 23:05:58
【问题描述】：

我想在没有 Pivot 函数的情况下使用 spark scala 转置下表

我使用的是 Spark 1.5.1，而 Pivot 功能在 1.5.1 中不支持。请提出合适的方法来转置下表：

Customer Day   Sales
1        Mon    12
1        Tue    10
1        Thu    15
1        Fri     2
2        Sun    10
2        Wed     5
2        Thu     4
2        Fri     3

输出表：

Customer Sun Mon Tue Wed Thu Fri
   1      0   12  10   0  15  2
   2     10    0   0   5  4   3

以下代码不起作用，因为我使用的是 Spark 1.5.1，并且可以从 Spark 1.6 获得数据透视函数：

    var Trans = Cust_Sales.groupBy("Customer").Pivot("Day").sum("Sales")

【问题讨论】：

标签： scala apache-spark apache-spark-sql apache-spark-1.5

【解决方案1】：

不确定效率如何，但您可以使用collect 获取所有不同的日期，然后添加这些列，然后使用groupBy 和sum：

// get distinct days from data (this assumes there are not too many of them):
val days: Array[String] = df.select("Day")
    .distinct()
    .collect()
    .map(_.getAs[String]("Day"))

// add column for each day with the Sale value if days match:
val withDayColumns = days.foldLeft(df) { 
    case (data, day) => data.selectExpr("*", s"IF(Day = '$day', Sales, 0) AS $day")
}

// wrap it up 
val result = withDayColumns
   .drop("Day")
   .drop("Sales")
   .groupBy("Customer")
   .sum(days: _*)

result.show()

哪个打印（几乎）你想要的：

+--------+--------+--------+--------+--------+--------+--------+
|Customer|sum(Tue)|sum(Thu)|sum(Sun)|sum(Fri)|sum(Mon)|sum(Wed)|
+--------+--------+--------+--------+--------+--------+--------+
|       1|      10|      15|       0|       2|      12|       0|
|       2|       0|       4|      10|       3|       0|       5|
+--------+--------+--------+--------+--------+--------+--------+

如果需要，我会留给您重命名/重新排序列。

【讨论】：

还有另一种解决方案，但它应用聚合或聚合ByKey，但我刚刚醒来，它需要大量的大脑体操来编写:)
@Tzach Zohar 非常感谢你，这真的很好。它在 18 分钟内处理了 3 GB 数据，在我的原始表中有 28 个不同的不同值（这里是天列）。再次感谢您的帮助
@Tzach Zohar，如果我在第二列（即本例中的 Day）中有大约 30,000 到 100,000 个不同的值来使用 scala 进行转置，还有其他选择吗？所以转置后我将有最多 100,000 列。如果您有大型数据集的解决方案，请告诉我。早期的解决方案在使用少量不同值时效果很好。

【解决方案2】：

如果您使用 python，下面的代码可能会有所帮助。假设您要转置 spark DataFrame df：

pandas_df = df.toPandas().transpose().reset_index()
transposed_df = sqlContext.createDataFrame(pandas_df)
transposed_df.show()

【讨论】：

【解决方案3】：

考虑一个有 6 列的数据框，我们希望按前 4 列分组并以 col5 为中心，同时在 col6 上进行聚合（比如求和）。因此，假设您不能使用 spark 1.6 版本，则可以将以下代码（在 spark 1.5 中）编写为：

val pivotedDf = df_to_pivot
    .groupBy(col1,col2,col3,col4)
    .pivot(col5)
    .agg(sum(col6))

这是具有相同输出但没有使用内置数据透视函数的代码：

import scala.collection.SortedMap

//Extracting the col5 distinct values to create the new columns
val distinctCol5Values = df_to_pivot
    .select(col(col5))
    .distinct
    .sort(col5)  // ensure that the output columns are in a consistent logical order
    .map(_.getString(0))
    .toArray
    .toSeq

//Grouping by the data frame to be pivoted on col1-col4
val pivotedAgg = df_to_pivot.rdd
      .groupBy{row=>(row.getString(col1Index),
      row.getDate(col2Index),
      row.getDate(col3Index),
      row.getString(col4Index))}

//Initializing a List of tuple of (String, double values) to be filled in the columns that will be created
val pivotColListTuple = distinctCol5Values.map(ft=> (ft,0.0))
// Using Sorted Map to ensure the order is maintained
var distinctCol5ValuesListMap = SortedMap(pivotColListTuple : _*)
//Pivoting the data on col5 by opening the grouped data
val pivotedRDD = pivotedAgg.map{groupedRow=>
    distinctCol5ValuesListMap = distinctCol5ValuesListMap.map(ft=> (ft._1,0.0))
     groupedRow._2.foreach{row=>
//Updating the distinctCol5ValuesListMap values to reflect the changes
//Change this part accordingly to what you want
       distinctCol5ValuesListMap = distinctCol5ValuesListMap.updated(row.getString(col5Index),
         distinctCol5ValuesListMap.getOrElse(row.getString(col5Index),0.0)+row.getDouble(col6Index))
     }        
    Row.fromSeq(Seq(groupedRow._1._1,groupedRow._1._2,groupedRow._1._3,groupedRow._1._4) ++ distinctCol5ValuesListMap.values.toSeq)
  }

//Consructing the structFields for new columns
val colTypesStruct = distinctCol5ValuesListMap.map(colName=>StructField(colName._1,DoubleType))
//Adding the first four column structFields with the new columns struct
val opStructType = StructType(Seq(StructField(col1Name,StringType),
                                     StructField(col2Name,DateType),
                                     StructField(col3Name,DateType),
                                     StructField(col4Name,StringType)) ++ colTypesStruct )

//Creating the final data frame
val pivotedDF = sqlContext.createDataFrame(pivotedRDD,opStructType)

【讨论】：