数据集上的 avg() 是否产生最有效的 RDD？

Question

据我了解，这是在 Spark 中计算平均值的最有效方法：Spark : Average of values instead of sum in reduceByKey using Scala。

我的问题是：如果我使用带有 groupby 的高级数据集，后跟 Spark 函数的 avg()，我会得到相同的 RDD 吗？我可以信任 Catalyst 还是应该使用低级 RDD？我的意思是，编写低级代码会比数据集产生更好的结果吗？

示例代码：

employees
  .groupBy($"employee")
  .agg(
    avg($"salary").as("avg_salary")
  )

对比：

employees
.mapValues(employee => (employee.salary, 1)) // map entry with a count of 1
.reduceByKey {
  case ((sumL, countL), (sumR, countR)) => 
    (sumL + sumR, countL + countR)
}
.mapValues { 
  case (sum , count) => sum / count 
}

Answer 1

我不认为这是一个非黑即白的问题。一般来说，如果你有一个 RDD，特别是如果它是一个 PairRDD，并且需要 RDD 中的结果，那么使用 reduceByKey 解决是有意义的。另一方面，给定一个 DataFrame，我建议使用 groupBy/agg(avg)。

需要考虑的几点：

内置优化

虽然reduceByKey 与 groupByKey 之类的函数相比相对有效，但它确实引入了stage 边界，因为该操作需要通过键对数据进行重新分区。根据 RDD 的分区，派生的stage 中的任务数量可能最终太少而无法利用可用的 cpu 内核，从而可能导致性能瓶颈。例如，可以通过在reduceByKey 中手动分配numPartition 来解决此类性能问题：

def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]

但关键是，要完全优化 RDD 操作，可能需要进行一些手动调整。相反，DataFrames 的大多数操作都是由内置的Catalyst 查询优化器自动优化的。

内存使用效率

也许要考虑的另一个更重要的因素与大型数据集的内存使用有关。当 RDD 需要跨节点分布或写入磁盘时，Spark 会将每一行数据序列化为对象，但会产生昂贵的垃圾收集开销。另一方面，通过了解 DataFrame 的模式，Spark 不需要将数据序列化为对象。 Tungsten 执行引擎可以利用堆外内存以二进制格式存储数据以进行转换，从而更有效地使用内存。

总之，虽然使用低级代码进行调整可能会有更多的旋钮，但由于优化不足、序列化的额外成本等原因，这并不一定会带来更高性能的代码。

Answer 2

我们可以从 Spark 生成的计划中得出结论。

这是DataFrame语法的计划-

    val employees = spark.createDataFrame(Seq(("E1",100.0), ("E2",200.0),("E3",300.0))).toDF("employee","salary")

    employees
      .groupBy($"employee")
      .agg(
        avg($"salary").as("avg_salary")
      ).explain(true)

计划-

== Parsed Logical Plan ==
'Aggregate ['employee], [unresolvedalias('employee, None), avg('salary) AS avg_salary#11]
+- Project [_1#0 AS employee#4, _2#1 AS salary#5]
   +- LocalRelation [_1#0, _2#1]

== Analyzed Logical Plan ==
employee: string, avg_salary: double
Aggregate [employee#4], [employee#4, avg(salary#5) AS avg_salary#11]
+- Project [_1#0 AS employee#4, _2#1 AS salary#5]
   +- LocalRelation [_1#0, _2#1]

== Optimized Logical Plan ==
Aggregate [employee#4], [employee#4, avg(salary#5) AS avg_salary#11]
+- LocalRelation [employee#4, salary#5]

== Physical Plan ==
*(2) HashAggregate(keys=[employee#4], functions=[avg(salary#5)], output=[employee#4, avg_salary#11])
+- Exchange hashpartitioning(employee#4, 10)
   +- *(1) HashAggregate(keys=[employee#4], functions=[partial_avg(salary#5)], output=[employee#4, sum#17, count#18L])
      +- LocalTableScan [employee#4, salary#5]

正如计划建议的那样，首先“HashAggregate”发生部分平均，然后“exchange hashpartitioning”发生完全平均。结论是催化剂优化了 DataFrame 操作，就好像我们使用“reduceByKey”语法编程一样。所以我们不必承担编写低级代码的负担。

这是 RDD 代码和计划的样子。

    employees
      .map(employee => ("key",(employee.getAs[Double]("salary"), 1))) // map entry with a count of 1
      .rdd.reduceByKey {
      case ((sumL, countL), (sumR, countR)) =>
        (sumL + sumR, countL + countR)
    }
    .mapValues {
      case (sum , count) => sum / count
    }.toDF().explain(true)

计划 -

== Parsed Logical Plan ==
SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]))._1, true, false) AS _1#30, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]))._2 AS _2#31]
+- ExternalRDD [obj#29]

== Analyzed Logical Plan ==
_1: string, _2: double
SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]))._1, true, false) AS _1#30, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]))._2 AS _2#31]
+- ExternalRDD [obj#29]

== Optimized Logical Plan ==
SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple2, true])._1, true, false) AS _1#30, assertnotnull(input[0, scala.Tuple2, true])._2 AS _2#31]
+- ExternalRDD [obj#29]

== Physical Plan ==
*(1) SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple2, true])._1, true, false) AS _1#30, assertnotnull(input[0, scala.Tuple2, true])._2 AS _2#31]
+- Scan[obj#29]

该计划经过优化，还涉及将数据序列化为对象，这意味着额外的内存压力。

结论

我会使用 daraframe 语法，因为它简单且性能可能更好。

Answer 3

在println("done")调试，转到http://localhost:4040/stages/，你会得到结果。

val spark = SparkSession
  .builder()
  .master("local[*]")
  .appName("example")
  .getOrCreate()

val employees = spark.createDataFrame(Seq(("employee1",1000),("employee2",2000),("employee3",1500))).toDF("employee","salary")
import spark.implicits._
import org.apache.spark.sql.functions._
// Spark functions
employees
  .groupBy("employee")
  .agg(
    avg($"salary").as("avg_salary")
  ).show()
// your low-level code

println("done")