title: Can't access broadcast variable in transformation
body: <p>I'm having problems accessing a variable from inside a transformation function.  Could someone help me out?
  Here are my relevant classes and functions.</p>
  
  <pre><code>@SerialVersionUID(889949215L)
  object MyCache extends Serializable {
      @transient lazy val logger = Logger(getClass.getName)
      @volatile var cache: Broadcast[Map[UUID, Definition]] = null
  
      def getInstance(sparkContext: SparkContext) : Broadcast[Map[UUID, Definition]] = {
          if (cache == null) {
              synchronized {
                  val map = sparkContext.cassandraTable("keyspace", "table")
                     .collect()
                     .map(m =&gt; m.getUUID("id") -&gt;
                          Definition(m.getString("c1"), m.getString("c2"), m.getString("c3"),
                                  m.getString("c4"))).toMap
                  cache = sparkContext.broadcast(map)
              }
          }
          cache
      }
  }
  </code></pre>
  
  <p>In a different file:</p>
  
  <pre><code>object Processor extends Serializable {
      @transient lazy val logger = Logger(getClass.getName)
  
      def processData[T: ClassTag](rawStream: DStream[(String, String)], ssc: StreamingContext,
                                          processor: (String, Broadcast[Map[UUID, Definition]]) =&gt; T): DStream[T] = {
          MYCache.getInstance(ssc.sparkContext)
          var newCacheValues = Map[UUID, Definition]()
          rawStream.cache()
          rawStream
            .transform(rdd =&gt; {
                  val array = rdd.collect()
                  array.foreach(r =&gt; {
                        val value = getNewCacheValue(r._2, rdd.context)
                        if (value.isDefined) {
                            newCacheValues = newCacheValues + value.get
                        }
                  })
                  rdd
             })
         if (newCacheValues.nonEmpty) {
             logger.info(s"Rebroadcasting.  There are ${newCacheValues.size} new values")
             logger.info("Destroying old cache")
             MyCache.cache.destroy()
             // this is probably wrong here, destroying object, but then referencing it.  But I haven't gotten to this part yet.
             MyCache.cache = ssc.sparkContext.broadcast(MyCache.cache.value ++ newCacheValues)
         }
         rawStream
            .map(r =&gt; {
                 println("######################")
                 println(MyCache.cache.value)
                 r
            })
            .map(r =&gt; processor(r._2, MyCache.cache.value))
            .filter(r =&gt; null != r)
     }
  }
  </code></pre>
  
  <p>Every time I run this I get <code>SparkException: Failed to get broadcast_1_piece0 of broadcast_1</code> when trying to access cache.value</p>
  
  <p>When I add a <code>println(MyCache.cache.values)</code> right after the <code>.getInstance</code> I'm able to access the broadcast variable, but when I deploy it to a mesos cluster I'm unable to access the broadcast values again, but with a null pointer exception.</p>
  
  <p><strong>Update:</strong></p>
  
  <p>The error I'm seeing is on <code>println(MyCache.cache.value)</code>.  I shouldn't have added this if statement containing the destroy, because my tests are never hitting that.  </p>
  
  <p>The basics of my application are, I have a table in cassandra that won't be updated very much.  But I need to do some validation on some streaming data.  So I want to pull all the data from this table, that isn't update much, into memory.  <code>getInstance</code> pulls the whole table in on startup, and then I check all my streaming data to see if I need to pull from cassandra again (which I will have to very rarely).  The transform and collect is where I check to see if I need to pull new data in.  But since there is a chance that my table will be updated, I will need to update the broadcast occasionally.  So my idea was to destroy it and then rebroadcast.  I will update that once I get the other stuff working.  </p>
  
  <p>I get the same error if I comment out the destroy and rebroadcast.</p>
  
  <p><strong>Another update:</strong></p>
  
  <p>I need to access the broadcast variable in <code>processor</code> this line: <code>.map(r =&gt; processor(r._2, MyCache.cache.value))</code>.</p>
  
  <p>I'm able to broadcast variable in the transform, and if I do <code>println(MyCache.cache.value)</code> in the transform, then all my tests pass, and I'm able to then access the broadcast in <code>processor</code></p>
  
  <p><strong>Update:</strong></p>
  
  <pre><code>rawStream
      .map(r =&gt; {
        println("$$$$$$$$$$$$$$$$$$$")
        println(metrics.value)
        r
      })
  </code></pre>
  
  <p>This is the stack trace I get when it hits this line.</p>
  
  <pre><code>    ERROR org.apache.spark.executor.Executor - Exception in task 0.0 in stage 135.0 (TID 114)
      java.io.IOException: org.apache.spark.SparkException: Failed to get broadcast_1_piece0 of broadcast_1
          at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1222)
          at org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:165)
          at org.apache.spark.broadcast.TorrentBroadcast._value$lzycompute(TorrentBroadcast.scala:64)
          at org.apache.spark.broadcast.TorrentBroadcast._value(TorrentBroadcast.scala:64)
          at org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:88)
          at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:70)
          at com.uptake.readings.ingestion.StreamProcessors$$anonfun$processIncomingKafkaData$4.apply(StreamProcessors.scala:160)
          at com.uptake.readings.ingestion.StreamProcessors$$anonfun$processIncomingKafkaData$4.apply(StreamProcessors.scala:158)
          at scala.collection.Iterator$$anon$11.next(Iterator.scala:370)
          at scala.collection.Iterator$$anon$11.next(Iterator.scala:370)
          at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:414)
          at org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:284)
          at org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:171)
          at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:78)
          at org.apache.spark.rdd.RDD.iterator(RDD.scala:268)
          at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
          at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
          at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)
          at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
          at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
          at org.apache.spark.scheduler.Task.run(Task.scala:89)
          at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
          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)
      Caused by: org.apache.spark.SparkException: Failed to get broadcast_1_piece0 of broadcast_1
          at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1$$anonfun$2.apply(TorrentBroadcast.scala:138)
          at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1$$anonfun$2.apply(TorrentBroadcast.scala:138)
          at scala.Option.getOrElse(Option.scala:121)
          at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1.apply$mcVI$sp(TorrentBroadcast.scala:137)
          at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1.apply(TorrentBroadcast.scala:120)
          at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1.apply(TorrentBroadcast.scala:120)
          at scala.collection.immutable.List.foreach(List.scala:381)
          at org.apache.spark.broadcast.TorrentBroadcast.org$apache$spark$broadcast$TorrentBroadcast$$readBlocks(TorrentBroadcast.scala:120)
          at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$readBroadcastBlock$1.apply(TorrentBroadcast.scala:175)
          at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1219)
          ... 24 more
  </code></pre>