Spark 广播报错：超过 spark.akka.frameSize 考虑使用广播

Question

我有一个名为“edges”的大数据

org.apache.spark.rdd.RDD[org.apache.spark.graphx.Edge[(String, Int)]] = MappedRDD[27] at map at <console>:52

当我在独立模式下工作时，我能够收集、计数和保存这个文件。现在，在集群上，我收到了这个错误

edges.count
...
Serialized task 28:0 was 12519797 bytes which exceeds spark.akka.frameSize
  (10485760 bytes). Consider using broadcast variables for large values.

与 .saveAsTextFile("edges") 相同

这是来自 spark-shell。我尝试过使用该选项
--driver-java-options "-Dspark.akka.frameSize=15"

但是当我这样做时，它只是无限期地挂起。任何帮助，将不胜感激。

** 编辑 **

我的独立模式是 Spark 1.1.0，我的集群是 Spark 1.0.1。

此外，当我对 RDD 进行计数、收集或另存为*时会发生挂起，但定义它或对其进行过滤器工作正常。

Answer 1

“考虑对大值使用广播变量”错误消息通常表明您在函数闭包中捕获了一些大变量。例如，你可能写过类似

val someBigObject = ...
rdd.mapPartitions { x => doSomething(someBigObject, x) }.count()

这会导致someBigObject 被捕获并与您的任务一起序列化。如果你正在做类似的事情，你可以使用 broadcast variable 代替，这将导致仅对对象的引用存储在任务本身中，而实际的对象数据将单独发送。

在 Spark 1.1.0+ 中，不必为此使用广播变量，因为任务将自动被广播（有关详细信息，请参阅 SPARK-2521）。仍然有使用广播变量的理由（例如在多个动作/作业之间共享一个大对象），但您不会需要使用它来避免帧大小错误。

另一种选择是增加 Akka 帧大小。在任何 Spark 版本中，您应该能够在创建 SparkContext 之前在 SparkConf 中设置 spark.akka.frameSize 设置。不过，您可能已经注意到，在为您创建上下文的spark-shell 中，这有点困难。在较新版本的 Spark（1.1.0 和更高版本）中，您可以在启动 spark-shell 时传递 --conf spark.akka.frameSize=16。在 Spark 1.0.1 或 1.0.2 中，您应该可以改为传递 --driver-java-options "-Dspark.akka.frameSize=16"。