Periodic Broadcast in Apache Spark Streaming

Question

I am implementing a stream learner for text classification. There are some single-valued parameters in my implementation that needs to be updated as new stream items arrive. For example, I want to change learning rate as the new predictions are made. However, I doubt that there is a way to broadcast variables after the initial broadcast. So what happens if I need to broadcast a variable every time I update it. If there is a way to do it or a workaround for what I want to accomplish in Spark Streaming, I'd be happy to hear about it.

Thanks in advance.

score 3 · Answer 1 · edited May 23 '17 at 10:29

I got this working by creating a wrapper class over the broadcast variable. The updateAndGet method of wrapper class returns the refreshed broadcast variable. I am calling this function inside dStream.transform -> as per the Spark Documentation

http://spark.apache.org/docs/latest/streaming-programming-guide.html#transform-operation

Transform Operation states: "the supplied function gets called in every batch interval. This allows you to do time-varying RDD operations, that is, RDD operations, number of partitions, broadcast variables, etc. can be changed between batches."

BroadcastWrapper class will look like :

public class BroadcastWrapper {
private Broadcast<ReferenceData> broadcastVar;
private Date lastUpdatedAt = Calendar.getInstance().getTime();

private static BroadcastWrapper obj = new BroadcastWrapper();

private BroadcastWrapper(){}

public static BroadcastWrapper getInstance() {
       return obj;
}

public JavaSparkContext getSparkContext(SparkContext sc) {
      JavaSparkContext jsc = JavaSparkContext.fromSparkContext(sc);
      return jsc;
}

public Broadcast<ReferenceData> updateAndGet(SparkContext sparkContext){
       Date currentDate = Calendar.getInstance().getTime();
       long diff = currentDate.getTime()-lastUpdatedAt.getTime();
       if (var == null || diff > 60000) { //Lets say we want to refresh every 1 min = 60000 ms
           if (var != null)
              var.unpersist();
           lastUpdatedAt = new Date(System.currentTimeMillis());

           //Your logic to refresh
           ReferenceData data = getRefData();

           var = getSparkContext(sparkContext).broadcast(data);
      }
      return var;
}
}

You can use this broadcast variable updateAndGet function in stream.transform method that allows RDD-RDD transformations

objectStream.transform(stream -> {

  Broadcast<Object> var = BroadcastWrapper.getInstance().updateAndGet(stream.context());

/**Your code to manipulate stream **/
});

Refer to my full answer from this pos :https://stackoverflow.com/a/41259333/3166245

Hope it helps

score 2 · Answer 2 · answered Feb 18 '15 at 06:34

2

My understanding is once a broadcast variable is initially sent out, it is 'read only'. I believe you can update the broadcast variable on the local nodes, but not on remote nodes.

May be you need to consider doing this 'outside Spark'. How about using a noSQL store (Cassandra ..etc) or even Memcache? You can then update the variable from one task and periodically check this store from other tasks?

answered Feb 18 '15 at 06:34

Sujee Maniyam

1,093
1
9
15

What is the difference between local and remote nodes? – Novemberland Nov 11 '18 at 20:11

score 1 · Answer 3 · answered Jan 21 '16 at 10:16

I got an ugly play, but it worked! We can find how to get a broadcast value from a broadcast object. https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala#L114 just by broadcast id.

so i periodically rebroadcast through the same broadcast id.

val broadcastFactory = new TorrentBroadcastFactory()
broadcastFactory.unbroadcast(BroadcastId, true, true)
// append some ids to initIds
val broadcastcontent = broadcastFactory.newBroadcast[.Set[String]](initIds, false, BroadcastId)

and i can get BroadcastId from the first broadcast value.

val ids = ssc.sparkContext.broadcast(initIds)
// broadcast id
val BroadcastId = broadcastIds.id

then worker use ids as a Broadcast Type as normal.

def func(record: Array[Byte], bc: Broadcast[Set[String]]) = ???

score 1 · Answer 4 · answered Jul 17 '16 at 13:44

1

bkc.unpersist(true)
bkc.destroy() 
bkc = sc.broadcast(tableResultMap) 
bkv = bkc.value

You may try this,I not guarantee whether effective

answered Jul 17 '16 at 13:44

mseaspring

11
3

score 0 · Answer 5 · answered Jul 03 '15 at 22:59

It is best that you collect the data to the driver and then broadcast them to all nodes.

Use Dstream # foreachRDD to collect the computed RDDs at the driver and once you know when you need to change learning rate, then use SparkContext#broadcast(value) to send the new value to all nodes.

I would expect the code to look something like the following:

dStreamContainingBroadcastValue.foreachRDD{ rdd => 
      val valueToBroadcast = rdd.collect()
      sc.broadcast(valueToBroadcast)
}

You may also find this thread useful, from the spark user mailing list. Let me know if that works.

So, how do you figure he can read the broadcast variable from that snippet? It kind of defeats its purpose to return `Unit`. — kareblak, Oct 06 '15 at 09:08

Periodic Broadcast in Apache Spark Streaming

5 Answers5