Spark - 对一列进行分组并查找其他列的平均值

Question

我有一些 4 列（c1、c2、c3 和 c4）的数据，并通过一些 scala 代码将其放入 RDD 中。

我想按 c1 进行分组/分组，并找到每个 c1 组中 c2 的平均值和 c3 以及 c4 的平均值。

我正在查看 RDD:reduceByKey，但我还没有设法准确理解它的使用方式。有一个更好的方法吗？我如何通过 Scala API 执行此操作？

Answer 1

你说你有一个DataFrame，所以你可能不应该使用RDDAPI（这通常效率较低，在这种情况下也可能不太直观） - 这是使用 API 的解决方案DataFrame：

import org.apache.spark.sql.functions._

val result = df.groupBy("c1").agg(mean("c2"), mean("c3"), mean("c4"))

result将是具有以下模式的 DataFrame（假设c1以字符串开头）：

root
 |-- c1: string (nullable = true)
 |-- avg(c2): double (nullable = true)
 |-- avg(c3): double (nullable = true)
 |-- avg(c4): double (nullable = true)

编辑：

如果列列表是动态的，您可以轻松地将这样的列表映射到相应“手段”的列表中，并使用该列表聚合 DF：

val colsToCompute = List("c2", "c3", "c4") // can be loaded dynamically
val means: Seq[Column] = colsToCompute.map(mean)
val result = df.groupBy("c1").agg(means.head, means.tail: _*)

为了完整起见 - 这是一个使用RDD API 的解决方案，但是：

• 它不太简洁
• 对于动态数量的列进行“泛化”要困难得多
• 它可能表现更差

实现可能稍微短一些，但也简单不了多少：

val rdd: RDD[(String, Int, Int, Int)] = ...

val result: RDD[(String, (Double, Double, Double))] = rdd
  .keyBy(_._1)
  .mapValues { case (k, v1, v2, v3) => (1, v1, v2, v3) } // add base for counter
  .reduceByKey { case ((a1, a2, a3, a4), (b1, b2, b3, b4)) => (a1+b1, a2+b2, a3+b3, a4+b4) } // sum counter and values
  .mapValues { case (count, v1, v2, v3) => (v1.toDouble/count, v2.toDouble/count, v3.toDouble/count) } // calculate means