gistfile1.scala

case class SetSizeAggregator(hllBits: Int, maxSetSize: Int = 10)
    extends MonoidAggregator[Array[Byte], Either[HLL, Set[Array[Byte]]], Long] {

  def prepare(in: Array[Byte]) = Right(Set(in))
  def present(sum: Either[HLL, Set[Array[Byte]]]) = {
    sum match {
      case Right(set) => set.size
      case Left(hll) => hll.approximateSize.estimate
    }
  }

  implicit val hllMonoid = new HyperLogLogMonoid(hllBits)
  val monoid = new EventuallyMonoid[HLL, Set[Array[Byte]]](hllMonoid.batchCreate(_))(_.size > maxSetSize)
}