Spark广播错误:超过spark.akka.frameSize考虑使用广播

Question

我有一个叫做"边缘"的大数据

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")相同

这是来自火花壳.我试过使用选项
--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您的任务被捕获和序列化.如果您正在执行类似的操作,则可以使用广播变量,这将导致仅将对象的引用存储在任务本身中,而实际的对象数据将单独发送.

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

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