Spark Streaming with CountMinSketch from Twitter Algebird

StreamingCMS.scala

import spark.streaming.{Seconds, StreamingContext}
import spark.storage.StorageLevel
import spark.streaming.examples.twitter.TwitterInputDStream
import com.twitter.algebird._
import spark.streaming.StreamingContext._
import spark.SparkContext._

/**
 * Example of using CountMinSketch monoid from Twitter's Algebird together with Spark Streaming's
 * TwitterInputDStream
 */
object StreamingCMS {
  def main(args: Array[String]) {
    if (args.length < 3) {
      System.err.println("Usage: TwitterStreamingCMS <master> <twitter_username> <twitter_password>" +
        " [filter1] [filter2] ... [filter n]")
      System.exit(1)
    }

    val DELTA = 1E-3
    val EPS = 0.01
    val SEED = 1
    val PERC = 0.001

    val Array(master, username, password) = args.slice(0, 3)
    val filters = args.slice(3, args.length)

    val ssc = new StreamingContext(master, "TwitterStreamingCMS", Seconds(10))
    val stream = new TwitterInputDStream(ssc, username, password, filters,
      StorageLevel.MEMORY_ONLY_SER)
    ssc.registerInputStream(stream)

    val users = stream.map(status => status.getUser.getId)

    var globalCMS = new CountMinSketchMonoid(DELTA, EPS, SEED, PERC).zero
    var globalExact = Map[Long, Int]()
    val mm = new MapMonoid[Long, Int]()

    val approxTopUsers = users.mapPartitions(ids => {
      val cms = new CountMinSketchMonoid(DELTA, EPS, SEED, PERC)
      ids.map(id => cms.create(id))
    }).reduce(_ ++ _)

    val exactTopUsers = users.map(id => (id, 1))
      .reduceByKey((a, b) => a + b)

    approxTopUsers.foreach(rdd => {
      if (rdd.count() != 0) {
        val partial = rdd.first()
        globalCMS ++= partial
        val globalTopK = globalCMS.heavyHitters.map(id => (id, globalCMS.frequency(id).estimate)).toSeq.sortBy(_._2).reverse.slice(0, 5)
        println("Approx heavy hitters at %2.2f%% users this batch: %s".format(PERC, partial.heavyHitters.mkString("[", ",", "]")))
        println("Approx heavy hitters at %2.2f%% users overall: %s".format(PERC, globalTopK.mkString("[", ",", "]")))
      }
    })

    exactTopUsers.foreach(rdd => {
      if (rdd.count() != 0) {
        val partialMap = rdd.collect().toMap
        val partialTopK = rdd.map({case (id, count) => (count, id)}).sortByKey(ascending = false).take(5)
        globalExact = mm.plus(globalExact.toMap, partialMap)
        val globalTopK = globalExact.toSeq.sortBy(_._2).reverse.slice(0, 5)
        println("Exact heavy hitters this batch: %s".format(partialTopK.mkString(",")))
        println("Exact heavy hitters overall: %s".format(globalTopK.mkString(",")))
      }
    })

    ssc.start()
  }
}

Hi MLnick, is there any advice for how to keep the value globalCMS safe in a long running spark streming job? I keep thinking I may lost it. And I can not figure out one way to save the cms value to a out store. Thanks.