+
+{% highlight scala %}
+
+object WordBlacklist {
+
+ @volatile private var instance: Broadcast[Seq[String]] = null
+
+ def getInstance(sc: SparkContext): Broadcast[Seq[String]] = {
+ if (instance == null) {
+ synchronized {
+ if (instance == null) {
+ val wordBlacklist = Seq("a", "b", "c")
+ instance = sc.broadcast(wordBlacklist)
+ }
+ }
+ }
+ instance
+ }
+}
+
+object DroppedWordsCounter {
+
+ @volatile private var instance: Accumulator[Long] = null
+
+ def getInstance(sc: SparkContext): Accumulator[Long] = {
+ if (instance == null) {
+ synchronized {
+ if (instance == null) {
+ instance = sc.accumulator(0L, "WordsInBlacklistCounter")
+ }
+ }
+ }
+ instance
+ }
+}
+
+wordCounts.foreachRDD((rdd: RDD[(String, Int)], time: Time) => {
+ // Get or register the blacklist Broadcast
+ val blacklist = WordBlacklist.getInstance(rdd.sparkContext)
+ // Get or register the droppedWordsCounter Accumulator
+ val droppedWordsCounter = DroppedWordsCounter.getInstance(rdd.sparkContext)
+ // Use blacklist to drop words and use droppedWordsCounter to count them
+ val counts = rdd.filter { case (word, count) =>
+ if (blacklist.value.contains(word)) {
+ droppedWordsCounter += count
+ false
+ } else {
+ true
+ }
+ }.collect()
+ val output = "Counts at time " + time + " " + counts
+})
+
+{% endhighlight %}
+
+See the full [source code]({{site.SPARK_GITHUB_URL}}/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala).
+
+
+{% highlight java %}
+
+class JavaWordBlacklist {
+
+ private static volatile Broadcast> instance = null;
+
+ public static Broadcast> getInstance(JavaSparkContext jsc) {
+ if (instance == null) {
+ synchronized (JavaWordBlacklist.class) {
+ if (instance == null) {
+ List wordBlacklist = Arrays.asList("a", "b", "c");
+ instance = jsc.broadcast(wordBlacklist);
+ }
+ }
+ }
+ return instance;
+ }
+}
+
+class JavaDroppedWordsCounter {
+
+ private static volatile Accumulator instance = null;
+
+ public static Accumulator getInstance(JavaSparkContext jsc) {
+ if (instance == null) {
+ synchronized (JavaDroppedWordsCounter.class) {
+ if (instance == null) {
+ instance = jsc.accumulator(0, "WordsInBlacklistCounter");
+ }
+ }
+ }
+ return instance;
+ }
+}
+
+wordCounts.foreachRDD(new Function2, Time, Void>() {
+ @Override
+ public Void call(JavaPairRDD rdd, Time time) throws IOException {
+ // Get or register the blacklist Broadcast
+ final Broadcast> blacklist = JavaWordBlacklist.getInstance(new JavaSparkContext(rdd.context()));
+ // Get or register the droppedWordsCounter Accumulator
+ final Accumulator droppedWordsCounter = JavaDroppedWordsCounter.getInstance(new JavaSparkContext(rdd.context()));
+ // Use blacklist to drop words and use droppedWordsCounter to count them
+ String counts = rdd.filter(new Function, Boolean>() {
+ @Override
+ public Boolean call(Tuple2 wordCount) throws Exception {
+ if (blacklist.value().contains(wordCount._1())) {
+ droppedWordsCounter.add(wordCount._2());
+ return false;
+ } else {
+ return true;
+ }
+ }
+ }).collect().toString();
+ String output = "Counts at time " + time + " " + counts;
+ }
+}
+
+{% endhighlight %}
+
+See the full [source code]({{site.SPARK_GITHUB_URL}}/blob/master/examples/src/main/java/org/apache/spark/examples/streaming/JavaRecoverableNetworkWordCount.java).
+
+
+{% highlight python %}
+
+def getWordBlacklist(sparkContext):
+ if ('wordBlacklist' not in globals()):
+ globals()['wordBlacklist'] = sparkContext.broadcast(["a", "b", "c"])
+ return globals()['wordBlacklist']
+
+def getDroppedWordsCounter(sparkContext):
+ if ('droppedWordsCounter' not in globals()):
+ globals()['droppedWordsCounter'] = sparkContext.accumulator(0)
+ return globals()['droppedWordsCounter']
+
+def echo(time, rdd):
+ # Get or register the blacklist Broadcast
+ blacklist = getWordBlacklist(rdd.context)
+ # Get or register the droppedWordsCounter Accumulator
+ droppedWordsCounter = getDroppedWordsCounter(rdd.context)
+
+ # Use blacklist to drop words and use droppedWordsCounter to count them
+ def filterFunc(wordCount):
+ if wordCount[0] in blacklist.value:
+ droppedWordsCounter.add(wordCount[1])
+ False
+ else:
+ True
+
+ counts = "Counts at time %s %s" % (time, rdd.filter(filterFunc).collect())
+
+wordCounts.foreachRDD(echo)
+
+{% endhighlight %}
+
+See the full [source code]({{site.SPARK_GITHUB_URL}}/blob/master/examples/src/main/python/streaming/recoverable_network_wordcount.py).
+
+
+