为什么当我使用扩展应用程序时，火花广播无法正常工作？ [英] Why spark broadcast doesn't work well when I use extends App?

问题描述

第一个代码引发空指针异常。

  object TryBroadcast扩展App {
 val conf = new SparkConf（ ）.setAppName（ o_o）
 val sc = new SparkContext（conf）
 val sample = sc.parallelize（1至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到1024）
 val bro = sc.broadcast（6666）
 val broSample = sample.map（x => x.toString + bro.value）
 broSample.collect（）。foreach（println）
} 
} 
  

scala版本：2.10.5
spa rk版本：1.4.0
stackTrace：

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

解决方案

bro 完全不同。在第一个实例中，它是单例类实例（ TryBroadcast ）上的一个字段。在第二个变量中，它是一个局部变量。

该局部变量被捕获，序列化并发送给执行者。在第一种情况下，引用是针对字段的，因此单例将被捕获并发送。我不确定Scala单例的构建方式和捕获方式。显然，在这种情况下，它在执行程序上访问时最终未初始化。

您可以将 bro 设为本地像这样的变量：

  object TryBroadcast扩展了应用{{b $ b val conf = new SparkConf（）。setAppName（ o_o ）
 val sc =新的SparkContext（conf）
 val sample = sc.parallelize（1至1024）
 val broSample = {
 val bro = sc.broadcast（6666）
 sample.map（x => x.toString + bro.value）
} 
 broSample.collect（）。foreach（println）
} 
  

The first code throws null pointer exception.

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)
}

The second works well.

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)
  }
}

It seems spark broadcast has something conflict with scala.App

scala version: 2.10.5 spark version: 1.4.0 stackTrace:

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)

解决方案

bro in the two cases is quite different. In the first one it's a field on a singleton class instance (TryBroadcast). In the second one it is a local variable.

I the local variable gets captured, serialized and sent over to the executors. In the first case the reference is to a field, so the singleton would get captured and sent. I'm not sure how a Scala singleton is built and how it is captured. Apparently in this case it ends up uninitialized when it is accessed on the executor.

You could make bro a local variable like this:

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)
}

这篇关于为什么当我使用扩展应用程序时，火花广播无法正常工作？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！