Scalding Algebird CMS

gistfile1.txt

import com.twitter.scalding._
import com.twitter.algebird._
 
/**
 * More sensible aggregation with Monoids.
 * Use SketchMap to get only the top words that we are interested about.
 * SketchMap is a generalization of the CountMinSketch in Algebird. Holds list of top items.
 * The size of the CMS will not grow so this will not run out of mem.
 */
class WordCount5(args: Args) extends Job(args)  {
  implicit def utf8(s: String): Array[Byte] = com.twitter.bijection.Injection.utf8(s)
  implicit val cmm = new SketchMapMonoid[String, Long](128, 6, 0, 20) // top 20
  type ApproxMap = SketchMap[String, Long]
 
  TextLine(args("input"))
    .flatMap('line -> 'word) { tokenize }
    .map('word -> 'word) { w: String => cmm.create((w, 1L)) }
    .groupAll{ _.sum[ApproxMap]('word) }
    .flatMap('word -> ('word, 'size)) { words: ApproxMap => words.heavyHitters }
    .write(Tsv(args("output")))
 
  def tokenize(text: String): Array[String] = {
    text.toLowerCase.replaceAll("[^a-z0-9\\s]", "").split("\\s+").filter( _ != "")
  }
}