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

Question

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

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

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

Answer 1

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

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