将 MapPartitionsRDD 转换为 DataFrame 并按 2 个键对数据进行分组答案

【问题标题】：Convert MapPartitionsRDD to DataFrame and grouping data by 2 keys将 MapPartitionsRDD 转换为 DataFrame 并按 2 个键对数据进行分组
【发布时间】：2020-06-10 03:59:18
【问题描述】：

我有一个如下所示的数据框：

  country | user | count
  ----------------------
  Germany | Sarah| 2
  China   | Paul | 1
  Germany | Alan | 3
  Germany | Paul | 1
          ...

我要做的是将此数据帧转换为另一个看起来像这样的数据帧：

  dimension | value
  --------------------------------------------
  Country   | [Germany -> 4, China -> 1]
  --------------------------------------------
  User      | [Sarah -> 2, Paul -> 2, Alan -> 3]
          ...

起初我尝试这样做：

  var newDF = Seq.empty[(String, Map[String,Long])].toDF("dimension", "value")
  df.collect()
    .foreach(row => { Array(0,1)
            .map(pos => 
             newDF = newDF.union(Seq((df.columns.toSeq(pos).toString, Map(row.mkString(",").split(",")(pos) -> row.mkString(",").split(",")(2).toLong))).toDF())
             )
     })
  val newDF2 = newDF.groupBy("dimension").agg(collect_list("value")).as[(String, Seq[Map[String, Long]])].map {case (id, list) => (id, list.reduce(_ |+| _))}.toDF("dimension", "value")

但是collect() 正在杀死我的司机。因此，我尝试这样做：

 class DimItem[T](val dimension: String, val value: String, val metric: T) 


 val items: RDD[DimItem[Long]] = df.rdd.flatMap(row => {
                                dims.zipWithIndex.map{case (dim, i) => 
                                                  new DimItem(dim, row(i).toString, row(13).asInstanceOf[Long])
                                                  }
                                })  
 // with the format [ DimItem(Country, Germany, 2), DimItem(User, Sarah, 2)], ...

val itemsGrouped: RDD[((String, String), Iterable[DimItem[Long]])] = items.groupBy(x => (x.dimension, x.value))
val aggregatedItems: RDD[DimItem[Long]] = itemsGrouped.map{case (key, items) => new DimItem(key._1, key._2, items.reduce((a,b) => a.metric + b.metric)}

这个想法是保存在一个 RDD 对象中，如 (Country, China, 1), (Country, Germany, 3), (Country, Germany, 1), ... 然后按 2 个第一个键 ( Country, China), (Country, Germany), ... 分组后，将它们的计数相加。例如：拥有 (Country, Germany, 3), (Country, Germany, 1) 将变为 (Country, Germany, 4)。

但是一旦我到达这里，它就会告诉我在items.reduce() 中存在不匹配：它需要一个 DimItem[Long] 但得到一个 Long。

下一步将按“维度”键对其进行分组，并在“值”列中创建Map[String, Int]()格式并将其转换为DF。

我有 2 个问题。

First：最后一段代码正确吗？

第二：如何将这个 MapPartitionsRDD 转换为 DF？

【问题讨论】：

完全删除 RDD 并仅使用与 RDD 相比至少具有两大优势的数据帧 API 怎么样？
嗨，Alexandros，我尝试使用 DataFrame API 完成所有操作，但不知何故，它没有将数据保存为我期望的列表映射结构。你能提供一些示例代码吗？我会按照您指出的再尝试一次，谢谢您的建议！
嗨@rach23 我添加了基于数据框 API 的建议方法。请让我知道它是否有效
谢谢亚历山德罗斯！有用。如果我们有 N 列而不是这 2 列（国家和用户），您会像示例中那样为每列创建一个数据框吗？只是想知道您对给定几列的最佳方法的看法。
那么只需为每一列调用下面的transform 函数并最终调用联合来附加它们

标签： scala apache-spark rdd

【解决方案1】：

这是一种基于数据框 API 的解决方案：

import org.apache.spark.sql.functions.{lit, map_from_arrays, collect_list}

def transform(df :DataFrame, colName: String) : DataFrame = 
  df.groupBy(colName)
    .agg{sum("count").as("sum")}
    .agg{
      map_from_arrays(
        collect_list(colName),
        collect_list("sum")
      ).as("value")
    }.select(lit(colName).as("dimension"), $"value")

val countryDf = transform(df, "country")
val userDf = transform(df, "user")

countryDf.unionByName(userDf).show(false)

// +---------+----------------------------------+
// |dimension|value                             |
// +---------+----------------------------------+
// |Country  |[Germany -> 6, China -> 1]        |
// |User     |[Sarah -> 2, Alan -> 3, Paul -> 2]|
// +---------+----------------------------------+

分析：首先我们分别得到国家和用户分组的总和。接下来，我们向管道添加一个自定义聚合，将先前的结果收集到一个地图中。地图将通过 Spark 2.4.0 中的 map_from_arrays 函数填充。我们使用 collect_list 收集地图的键/值。最后，我们联合这两个数据框来填充最终结果。

【讨论】：