Merge pull request #194 from stephanh/more-aggregators

Added Aggregators for BloomFilter, CountMinSketch, MapSketch, HyperLogLog, Moments, Min, Max

Author: johnynek

algebird-core/src/main/scala/com/twitter/algebird/BloomFilter.scala

@@ -338,3 +338,13 @@ case class BFHash(numHashes: Int, width: Int, seed: Long = 0L) extends Function1
   }
 }
 
+case class BloomFilterAggregator(bfMonoid: BloomFilterMonoid)  extends MonoidAggregator[String, BF, BF] {
+  val monoid = bfMonoid
+
+  def prepare(value: String) = monoid.create(value)
+  def present(bf: BF) = bf
+}
+
+object BloomFilterAggregator {
+  def apply(numHashes: Int, width: Int, seed: Int = 0): BloomFilterAggregator = BloomFilterAggregator(BloomFilterMonoid(numHashes, width, seed))
+}

algebird-core/src/main/scala/com/twitter/algebird/CountMinSketch.scala

@@ -111,12 +111,22 @@ class CountMinSketchMonoid(eps : Double, delta : Double, seed : Int,
 }
 
 object CMS {
-  def monoid(eps : Double, delta : Double, seed : Int, heavyHittersPct : Double = 0.01) =
+  def monoid(eps : Double, delta : Double, seed : Int, heavyHittersPct : Double = 0.01): CountMinSketchMonoid =
     new CountMinSketchMonoid(eps, delta, seed, heavyHittersPct)
 
-  def monoid(depth : Int, width : Int, seed : Int, heavyHittersPct : Double) =
+  def monoid(depth : Int, width : Int, seed : Int, heavyHittersPct : Double): CountMinSketchMonoid =
     new CountMinSketchMonoid(CMS.eps(width), CMS.delta(depth), seed, heavyHittersPct)
 
+  def aggregator(eps: Double, delta: Double, seed: Int, heavyHittersPct: Double = 0.01): CountMinSketchAggregator = {
+    val monoid = new CountMinSketchMonoid(eps, delta, seed, heavyHittersPct)
+    new CountMinSketchAggregator(monoid)
+  }
+
+  def aggregator(depth : Int, width : Int, seed : Int, heavyHittersPct : Double): CountMinSketchAggregator = {
+    val monoid = new CountMinSketchMonoid(CMS.eps(width), CMS.delta(depth), seed, heavyHittersPct)
+    new CountMinSketchAggregator(monoid)
+  }
+
   /**
    * Functions to translate between (eps, delta) and (depth, width). The translation is:
    * depth = ceil(ln 1/delta)
@@ -453,3 +463,14 @@ case class HeavyHitter(item : Long, count : Long)
 object HeavyHitter {
   val ordering = Ordering.by { hh : HeavyHitter => (hh.count, hh.item) }
 }
+
+/**
+  * An Aggregator for the CountMinSketch.
+  * Can be created using CMS.aggregator
+  */
+case class CountMinSketchAggregator(cmsMonoid : CountMinSketchMonoid) extends MonoidAggregator[Long, CMS, CMS] {
+  val monoid = cmsMonoid
+
+  def prepare(value : Long) = monoid.create(value)
+  def present(cms : CMS) = cms
+}

algebird-core/src/main/scala/com/twitter/algebird/HyperLogLog.scala

@@ -358,3 +358,21 @@ class HyperLogLogMonoid(val bits : Int) extends Monoid[HLL] {
     .getOrElse(Approximate.exact(0L)) //Empty lists have no intersection
   }
 }
+
+object HyperLogLogAggregator {
+  def apply(bits: Int): HyperLogLogAggregator = {
+    val monoid = new HyperLogLogMonoid(bits)
+    new HyperLogLogAggregator(monoid)
+  }
+
+  def sizeAggregator(bits: Int): Aggregator[Array[Byte], HLL, Double] = {
+    apply(bits).andThenPresent(_.estimatedSize)
+  }
+}
+
+case class HyperLogLogAggregator(val hllMonoid: HyperLogLogMonoid) extends MonoidAggregator[Array[Byte], HLL, HLL] {
+  val monoid = hllMonoid
+
+  def prepare(value: Array[Byte]) = monoid.create(value)
+  def present(hll: HLL) = hll
+}

algebird-core/src/main/scala/com/twitter/algebird/MomentsGroup.scala

@@ -41,7 +41,8 @@ case class Moments(m0 : Long, m1 : Double, m2 : Double, m3 : Double, m4 : Double
 
 object Moments {
   implicit val group = MomentsGroup
-  
+  implicit val aggregator = MomentsAggregator
+
   // Create a Moments object given a single value. This is useful for
   // initializing moment calculations at the start of a stream.
   def apply[V <% Double](value : V) = new Moments(1L, value, 0, 0, 0)
@@ -126,3 +127,10 @@ object MomentsGroup extends Group[Moments] {
     }
   }
 }
+
+object MomentsAggregator extends MonoidAggregator[Double, Moments, Moments] {
+  val monoid = MomentsGroup
+
+  def prepare(input: Double): Moments = Moments(input)
+  def present(m: Moments) = m
+}

algebird-core/src/main/scala/com/twitter/algebird/OrderedSemigroup.scala

@@ -28,6 +28,8 @@ object Max {
   def monoid[T](zero: => T)(implicit ord: Ordering[T]): Monoid[Max[T]] =
      Monoid.from(Max(zero)) { (l,r) => if(ord.gteq(l.get, r.get)) l else r }
 
+  def aggregator[T](implicit ord:Ordering[T]): MaxAggregator[T] = MaxAggregator()(ord)
+
   implicit def intMonoid: Monoid[Max[Int]] = monoid(Int.MinValue)
   implicit def longMonoid: Monoid[Max[Long]] = monoid(Long.MinValue)
   implicit def doubleMonoid: Monoid[Max[Double]] = monoid(Double.MinValue)
@@ -61,6 +63,8 @@ object Min {
   def monoid[T](zero: => T)(implicit ord: Ordering[T]): Monoid[Min[T]] =
      Monoid.from(Min(zero)) { (l,r) => if(ord.lteq(l.get, r.get)) l else r }
 
+  def aggregator[T](implicit ord:Ordering[T]): MinAggregator[T] = MinAggregator()(ord)
+
   implicit def intMonoid: Monoid[Min[Int]] = monoid(Int.MaxValue)
   implicit def longMonoid: Monoid[Min[Long]] = monoid(Long.MaxValue)
   implicit def doubleMonoid: Monoid[Min[Double]] = monoid(Double.MaxValue)
@@ -78,3 +82,15 @@ case class Last[@specialized(Int,Long,Float,Double) +T](get: T)
 object Last {
   implicit def semigroup[T] = Semigroup.from[Last[T]] { (l,r) => r }
 }
+
+case class MinAggregator[T](implicit ord: Ordering[T]) extends Aggregator[T, T, T] {
+  def prepare(v: T) = v
+  def reduce(l: T, r: T) = ord.min(l , r)
+  def present(v: T) = v
+}
+
+case class MaxAggregator[T](implicit ord: Ordering[T]) extends Aggregator[T, T, T] {
+  def prepare(v: T) = v
+  def reduce(l: T, r: T) = ord.max(l , r)
+  def present(v: T) = v
+}

algebird-core/src/main/scala/com/twitter/algebird/SketchMap.scala

@@ -128,6 +128,11 @@ object SketchMap {
                   (implicit serialization: K => Array[Byte], valueOrdering: Ordering[V], monoid: Monoid[V]): SketchMapMonoid[K, V] = {
     new SketchMapMonoid(width(eps), depth(delta), seed, heavyHittersCount)(serialization, valueOrdering, monoid)
   }
+
+  def aggregator[K, V](eps: Double, delta: Double, seed: Int, heavyHittersCount: Int)
+                      (implicit serialization: K => Array[Byte], valueOrdering: Ordering[V], monoid: Monoid[V]): SketchMapAggregator[K, V] = {
+    SketchMapAggregator(SketchMap.monoid(eps, delta, seed, heavyHittersCount))
+  }
 }
 
 case class SketchMap[K, V](
@@ -225,3 +230,13 @@ case class SketchMap[K, V](
   }
 }
 
+/**
+  * An Aggregator for the SketchMap.
+  * Can be created using SketchMap.aggregator
+  */
+case class SketchMapAggregator[K, V](skmMonoid : SketchMapMonoid[K, V]) extends MonoidAggregator[(K, V), SketchMap[K, V], SketchMap[K, V]] {
+  val monoid = skmMonoid
+
+  def prepare(value: (K,V)) = monoid.create(value)
+  def present(skm: SketchMap[K, V]) = skm
+}

algebird-test/src/test/scala/com/twitter/algebird/BloomFilterTest.scala

@@ -84,5 +84,20 @@ class BloomFilterTest extends Specification {
         size.max must be_>=(size.estimate)
       }
     }
+
+    "work as an Aggregator" in {
+      (0 to 10).foreach{
+        _ => {
+          val aggregator = BloomFilterAggregator(RAND.nextInt(5)+1, RAND.nextInt(64)+32, SEED)
+          val numEntries = 5
+          val entries = (0 until numEntries).map(_ => RAND.nextInt.toString)
+          val bf = aggregator(entries)
+
+          entries.foreach{
+            i => bf.contains(i.toString).isTrue must be_==(true)
+          }
+        }
+      }
+    }
   }
 }

algebird-test/src/test/scala/com/twitter/algebird/CountMinSketchTest.scala

@@ -196,5 +196,18 @@ class CountMinSketchTest extends Specification {
       cms3.heavyHitters must be_==(Set(5L))
       cms4.heavyHitters must be_==(Set[Long]())
     }
+
+    "work as an Aggregator" in {
+      val data1 = Seq(1L, 2L, 2L, 3L, 3L, 3L, 4L, 4L, 4L, 4L, 5L, 5L, 5L, 5L, 5L)
+      val cms1 = CMS.aggregator(EPS, DELTA, SEED, 0.01).apply(data1)
+      val cms2 = CMS.aggregator(EPS, DELTA, SEED, 0.1).apply(data1)
+      val cms3 = CMS.aggregator(EPS, DELTA, SEED, 0.3).apply(data1)
+      val cms4 = CMS.aggregator(EPS, DELTA, SEED, 0.9).apply(data1)
+
+      cms1.heavyHitters must be_==(Set(1L, 2L, 3L, 4L, 5L))
+      cms2.heavyHitters must be_==(Set(2L, 3L, 4L, 5L))
+      cms3.heavyHitters must be_==(Set(5L))
+      cms4.heavyHitters must be_==(Set[Long]())
+    }
   }
 }

algebird-test/src/test/scala/com/twitter/algebird/HyperLogLogTest.scala

@@ -146,6 +146,28 @@ class HyperLogLogTest extends Specification {
       }
     }
 
+    "work as an Aggregator and return a HLL" in {
+      List(5, 7, 10).foreach(bits => {
+        val aggregator = HyperLogLogAggregator(bits)
+        val data = (0 to 10000).map { i => r.nextInt(1000) }
+        val exact = exactCount(data).toDouble
+
+        val approxCount = aggregator(data.map(int2Bytes(_))).approximateSize.estimate.toDouble
+        scala.math.abs(exact - approxCount) / exact must be_<(3.5 * aveErrorOf(bits))
+      })
+    }
+
+    "work as an Aggregator and return size" in {
+      List(5, 7, 10).foreach(bits => {
+        val aggregator = HyperLogLogAggregator.sizeAggregator(bits)
+        val data = (0 to 10000).map { i => r.nextInt(1000) }
+        val exact = exactCount(data).toDouble
+
+        val estimate = aggregator(data.map(int2Bytes(_)))
+        scala.math.abs(exact - estimate) / exact must be_<(3.5 * aveErrorOf(bits))
+      })
+    }
+
     def verifySerialization(h : HLL) {
       fromBytes(toBytes(h)) must be_==(h)
       fromByteBuffer(java.nio.ByteBuffer.wrap(toBytes(h))) must be_==(h)

algebird-test/src/test/scala/com/twitter/algebird/MinMaxAggregatorTest.scala

@@ -0,0 +1,43 @@
+package com.twitter.algebird
+
+import org.specs._
+
+class MinMaxAggregatorTest extends Specification {
+  val data = List(1, 3, 5, 0, 7, 6)
+
+  sealed trait TestElementParent
+  case object TestElementA extends TestElementParent
+  case object TestElementB extends TestElementParent
+  case object TestElementC extends TestElementParent
+
+  implicit val testOrdering = Ordering.fromLessThan[TestElementParent]((x, y) => (x, y) match {
+    case (TestElementA, TestElementA) => false
+    case (TestElementA, _) => true
+    case (TestElementB, TestElementB) => false
+    case (TestElementB, TestElementA) => false
+    case (TestElementB, TestElementC) => true
+    case (TestElementC, _) => false
+  })
+
+  val data2 = List(TestElementC, TestElementA, TestElementB)
+
+  "MinAggregator" should {
+    "produce the minimum value" in {
+      val agg = Min.aggregator[Int]
+      agg(data) must be_==(0)
+
+      val agg2 = Min.aggregator[TestElementParent]
+      agg2(data2) must be_==(TestElementA)
+    }
+  }
+
+  "MaxAggregator" should {
+    "produce the maximum value" in {
+      val agg = Max.aggregator[Int]
+      agg(data) must be_==(7)
+
+      val agg2 = Max.aggregator[TestElementParent]
+      agg2(data2) must be_==(TestElementC)
+    }
+  }
+}

algebird-test/src/test/scala/com/twitter/algebird/MomentsGroupTest.scala

@@ -60,5 +60,23 @@ class MomentsGroupTest extends Specification {
 
     val m2 = getMoments(List(1, 1, 1, 2, 3))
     testApproxEq(m2.kurtosis, -0.921875)
-  }  
+  }
+
+  "Moments can be aggregated" in {
+    val m1 = MomentsAggregator(List(1, 2, 3, 4, 5))
+    testApproxEq(m1.count, 5)
+    testApproxEq(m1.mean , 3)
+    testApproxEq(m1.variance, 2)
+    testApproxEq(m1.skewness, 0)
+    testApproxEq(m1.kurtosis, -1.3)
+
+    val m2 = MomentsAggregator(List(1, 1, 1, 2, 3))
+    testApproxEq(m2.count, 5)
+    testApproxEq(m2.mean , 1.6)
+    testApproxEq(m2.variance, 0.64)
+    testApproxEq(m2.skewness, 0.84375)
+    testApproxEq(m2.kurtosis, -0.921875)
+
+
+  }
 }

algebird-test/src/test/scala/com/twitter/algebird/SketchMapTest.scala

@@ -130,5 +130,19 @@ class SketchMapTest extends Specification {
       val sm4 = monoid.plus(sm3, monoid.create(Seq((100, 100L), (200, 30L), (200, 20L), (200, 10L))))
       sm4.heavyHitters must be_==(List((100, 5L), (200, 10L)))
     }
+
+    "work as an Aggregator" in {
+      val data = Seq((1, 1L), (2, 2L), (3, 3L), (4, 4L), (5, 5L))
+
+      val sm1 = SketchMap.aggregator[Int, Long](EPS, DELTA, SEED, 5).apply(data)
+      val sm2 = SketchMap.aggregator[Int, Long](EPS, DELTA, SEED, 3).apply(data)
+      val sm3 = SketchMap.aggregator[Int, Long](EPS, DELTA, SEED, 1).apply(data)
+      val sm4 = SketchMap.aggregator[Int, Long](EPS, DELTA, SEED, 0).apply(data)
+
+      sm1.heavyHitterKeys must be_==(List(5, 4, 3, 2, 1))
+      sm2.heavyHitterKeys must be_==(List(5, 4, 3))
+      sm3.heavyHitterKeys must be_==(List(5))
+      sm4.heavyHitterKeys must be_==(List.empty[Int])
+    }
   }
 }