gistfile1.scala

def takeBy(pipe: TypedPipe[(K,V)], max: Int)(fn: V => Double): TypedPipe[(K,V)] = {
  implicit val qtreeSemi = QTreeSemigroup(4) //magic number, determines how much RAM the trees take 
  val qtrees = pipe.map{case (k,v) => k -> QTree(fn(v))}.sumByKey
  val maxV = qtrees.flatMap{case (k,q) =>
    if(q.size > max) {
      val targetQuantile = max.toDouble / q.size
      val (lower, upper) = q.quantileBounds(targetQuantile)
      Some(k -> upper) //this will give us at least max values; use lower to get at most max values
    }
    else
      None
  }
  pipe
    .hashLeftJoin(maxV)
    .flatMap{case (k,v,optMaxV) =>
      if(!optMaxV.isDefined || v <= optMaxV.get)
        Some(k -> v)
      else
        None
    }
}