scala - 为什么spark broadcast在我用extends App的时候效果不好？

Question

第一段代码抛出空指针异常。

object TryBroadcast extends App{
  val conf = new SparkConf().setAppName("o_o")
  val sc = new SparkContext(conf)
  val sample = sc.parallelize(1 to 1024)
  val bro = sc.broadcast(6666)
  val broSample = sample.map(x => x.toString + bro.value)
  broSample.collect().foreach(println)
}

第二种效果很好。

object TryBroadcast {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("o_o")
    val sc = new SparkContext(conf)
    val sample = sc.parallelize(1 to 1024)
    val bro = sc.broadcast(6666)
    val broSample = sample.map(x => x.toString + bro.value)
    broSample.collect().foreach(println)
  }
}

spark broadcast好像和scala.App有冲突

scala 版本:2.10.5 Spark 版本:1.4.0 堆栈跟踪:

lang.NullPointerException
    at TryBroadcast$$anonfun$1.apply(TryBroadcast.scala:11)
    at TryBroadcast$$anonfun$1.apply(TryBroadcast.scala:11)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
    at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
    at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
    at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
    at scala.collection.AbstractIterator.to(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
    at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
    at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
    at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:885)
    at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:885)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1765)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1765)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:63)
    at org.apache.spark.scheduler.Task.run(Task.scala:70)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

Answer 1

bro 在这两种情况下是完全不同的。在第一个中，它是单例类实例 (TryBroadcast) 上的一个字段。在第二个中，它是一个局部变量。

I 局部变量被捕获、序列化并发送给执行程序。在第一种情况下，引用是一个字段，因此单例将被捕获并发送。我不确定如何构建 Scala 单例以及如何捕获它。显然，在这种情况下，当在执行程序上访问它时，它最终未初始化。

你可以像这样使 bro 成为局部变量:

object TryBroadcast extends App {
  val conf = new SparkConf().setAppName("o_o")
  val sc = new SparkContext(conf)
  val sample = sc.parallelize(1 to 1024)
  val broSample = {
    val bro = sc.broadcast(6666)
    sample.map(x => x.toString + bro.value)
  }
  broSample.collect().foreach(println)
}