Spark 中的 accumulator 和 collect() 之间的性能差异是什么？

Question

Accumulator 基本上是 spark 中的共享变量，由执行程序更新，但只能由驱动程序读取。 spark中的Collect()是将所有数据从executors中获取到驱动中。

所以，当我最终只在驱动程序中获取数据时。那么，当我们使用累加器或collect() 将大 RDD 转换为 LIST 时，性能有何不同？

使用累加器将数据帧转换为列表的代码

val queryOutput = spark.sql(query)
val acc = spark.sparkContext.collectionAccumulator[Map[String,Any]]("JsonCollector")
val jsonString = queryOutput.foreach(a=>acc.add(convertRowToJSON(a)))
acc.value.asScala.toList


def convertRowToJSON(row: Row): Map[String,Any] = {
    val m = row.getValuesMap(row.schema.fieldNames)
    println(m)
    JSONObject(m).obj
  }

使用 collect() 将数据框转换为列表的代码

val queryOutput = spark.sql(query)
queryOutput.toJSON.collectAsList()

Answer 1

将大RDD转换为LIST

这不是一个好主意。 collect 会将数据从所有执行程序移动到驱动程序内存。如果内存不足，则会抛出 Out Of Memory (OOM) 异常。如果您的数据适合单机内存，那么您可能不需要 spark。

Spark 原生支持数字类型的累加器，程序员可以添加对新类型的支持。它们可用于实现计数器（如在 MapReduce 中）或求和。累加器的 OUT 参数应该是可以原子读取（例如 Int、Long）或线程安全（例如同步集合）的类型，因为它将从其他线程读取。

CollectionAccumulator .value 返回 List（ArrayList 实现），如果 size 大于驱动内存，它会抛出 OOM。