Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

SPARK-1941: Update streamlib to 2.7.0 and use HyperLogLogPlus instead of HyperLogLog. #897

Closed
wants to merge 15 commits into from
Closed

SPARK-1941: Update streamlib to 2.7.0 and use HyperLogLogPlus instead of HyperLogLog. #897

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
c0ef0c2
SPARK-1941: Update streamlib to 2.7.0 and use HyperLogLogPlus instead…
rxin May 28, 2014
e7786cb
Merge branch 'master' into hll
rxin May 31, 2014
1294be6
Updated HLL+.
rxin May 31, 2014
88cfe77
Updated documentation and restored the old incorrect API to maintain …
rxin Jun 2, 2014
9221b27
Merge branch 'master' into hll
rxin Jun 2, 2014
1db1522
Excluded util.SerializableHyperLogLog from MIMA check.
rxin Jun 2, 2014
6555bfe
Added a default method and re-arranged MimaExcludes.
rxin Jun 2, 2014
acaa524
Added the right exclude rules in MimaExcludes.
rxin Jun 2, 2014
354deb8
Added comment on the Mima exclude rules.
rxin Jun 2, 2014
e110d70
Merge branch 'master' into hll
rxin Jun 3, 2014
9e320c8
Incorporate code review feedback.
rxin Jun 3, 2014
41e649a
Update final mima list.
rxin Jun 3, 2014
e367527
One more round of code review.
rxin Jun 4, 2014
f154ea0
Added a comment on the value bound for testing.
rxin Jun 4, 2014
4d83f41
New error bound and non-randomness.
rxin Jun 4, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 77 additions & 13 deletions core/src/main/scala/org/apache/spark/api/java/JavaPairRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -672,38 +672,102 @@ class JavaPairRDD[K, V](val rdd: RDD[(K, V)])

/**
* Return approximate number of distinct values for each key in this RDD.
*
* The accuracy of approximation can be controlled through the relative standard deviation
* (relativeSD) parameter, which also controls the amount of memory used. Lower values result in
* more accurate counts but increase the memory footprint and vise versa. Uses the provided
* Partitioner to partition the output RDD.
*
* @param p The precision value for the normal set.
* <code>p</code> must be a value between 4 and <code>sp</code> (32 max).
* @param sp The precision value for the sparse set, between 0 and 32.
* If <code>sp</code> equals 0, the sparse representation is skipped.
* @param partitioner Partitioner to use for the resulting RDD.
*/
def countApproxDistinctByKey(relativeSD: Double, partitioner: Partitioner): JavaRDD[(K, Long)] = {
rdd.countApproxDistinctByKey(relativeSD, partitioner)
def countApproxDistinctByKey(p: Int, sp: Int, partitioner: Partitioner): JavaPairRDD[K, Long] = {
fromRDD(rdd.countApproxDistinctByKey(p, sp, partitioner))
}

/**
* Return approximate number of distinct values for each key this RDD.
* Return approximate number of distinct values for each key in this RDD.
*
* The accuracy of approximation can be controlled through the relative standard deviation
* (relativeSD) parameter, which also controls the amount of memory used. Lower values result in
* more accurate counts but increase the memory footprint and vise versa. The default value of
* relativeSD is 0.05. Hash-partitions the output RDD using the existing partitioner/parallelism
* level.
* more accurate counts but increase the memory footprint and vise versa. Uses the provided
* Partitioner to partition the output RDD.
*
* @param p The precision value for the normal set.
* <code>p</code> must be a value between 4 and <code>sp</code> (32 max).
* @param sp The precision value for the sparse set, between 0 and 32.
* If <code>sp</code> equals 0, the sparse representation is skipped.
* @param numPartitions The number of partitions in the resulting RDD.
*/
def countApproxDistinctByKey(relativeSD: Double = 0.05): JavaRDD[(K, Long)] = {
rdd.countApproxDistinctByKey(relativeSD)
def countApproxDistinctByKey(p: Int, sp: Int, numPartitions: Int): JavaPairRDD[K, Long] = {
fromRDD(rdd.countApproxDistinctByKey(p, sp, numPartitions))
}


/**
* Return approximate number of distinct values for each key in this RDD.
*
* The accuracy of approximation can be controlled through the relative standard deviation
* (relativeSD) parameter, which also controls the amount of memory used. Lower values result in
* more accurate counts but increase the memory footprint and vise versa. Uses the provided
* Partitioner to partition the output RDD.
*
* @param p The precision value for the normal set.
* <code>p</code> must be a value between 4 and <code>sp</code> (32 max).
* @param sp The precision value for the sparse set, between 0 and 32.
* If <code>sp</code> equals 0, the sparse representation is skipped.
*/
def countApproxDistinctByKey(p: Int, sp: Int): JavaPairRDD[K, Long] = {
fromRDD(rdd.countApproxDistinctByKey(p, sp))
}

/**
* Return approximate number of distinct values for each key in this RDD. This is deprecated.
* Use the variant with <code>p</code> and <code>sp</code> parameters instead.
*
* The accuracy of approximation can be controlled through the relative standard deviation
* (relativeSD) parameter, which also controls the amount of memory used. Lower values result in
* more accurate counts but increase the memory footprint and vise versa. HashPartitions the
* output RDD into numPartitions.
* more accurate counts but increase the memory footprint and vise versa. Uses the provided
* Partitioner to partition the output RDD.
*/
@Deprecated
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to provide some migration tips. Is there a mapping from relativeSD to precision numbers? Actually, relativeSD is much easier for users to understand.

def countApproxDistinctByKey(relativeSD: Double, partitioner: Partitioner): JavaPairRDD[K, Long] =
{
fromRDD(rdd.countApproxDistinctByKey(relativeSD, partitioner))
}

/**
* Return approximate number of distinct values for each key in this RDD. This is deprecated.
* Use the variant with <code>p</code> and <code>sp</code> parameters instead.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*
* @param relativeSD The relative standard deviation for the counter.
* Smaller values create counters that require more space.
*/
@Deprecated
def countApproxDistinctByKey(relativeSD: Double, numPartitions: Int): JavaPairRDD[K, Long] = {
fromRDD(rdd.countApproxDistinctByKey(relativeSD, numPartitions))
}

/**
* Return approximate number of distinct values for each key in this RDD. This is deprecated.
* Use the variant with <code>p</code> and <code>sp</code> parameters instead.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*
* @param relativeSD The relative standard deviation for the counter.
* Smaller values create counters that require more space.
*/
def countApproxDistinctByKey(relativeSD: Double, numPartitions: Int): JavaRDD[(K, Long)] = {
rdd.countApproxDistinctByKey(relativeSD, numPartitions)
@Deprecated
def countApproxDistinctByKey(relativeSD: Double): JavaPairRDD[K, Long] = {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note that I changed the return type from JavaRDD[(K, Long)] to JavaPairRDD[K, Long], because that is what it should've been.

However, in order to maintain complete API stability, I can change it back and just deprecated the old methods. The new methods certainly should return JavaPairRDD.

fromRDD(rdd.countApproxDistinctByKey(relativeSD))
}

/** Assign a name to this RDD */
Expand Down
27 changes: 26 additions & 1 deletion core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -564,8 +564,33 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
* (relativeSD) parameter, which also controls the amount of memory used. Lower values result in
* more accurate counts but increase the memory footprint and vise versa. The default value of
* relativeSD is 0.05.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*
* @param p The precision value for the normal set.
* <code>p</code> must be a value between 4 and <code>sp</code>.
* @param sp The precision value for the sparse set, between 0 and 32.
* If <code>sp</code> equals 0, the sparse representation is skipped.
*/
def countApproxDistinct(p: Int, sp: Int): Long = rdd.countApproxDistinct(p, sp)

/**
* Return approximate number of distinct elements in the RDD. This is deprecated. Use the
* variant with <code>p</code> and <code>sp</code> parameters instead.
*
* The accuracy of approximation can be controlled through the relative standard deviation
* (relativeSD) parameter, which also controls the amount of memory used. Lower values result in
* more accurate counts but increase the memory footprint and vise versa. The default value of
* relativeSD is 0.05.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*/
def countApproxDistinct(relativeSD: Double = 0.05): Long = rdd.countApproxDistinct(relativeSD)
@Deprecated
def countApproxDistinct(relativeSD: Double): Long = rdd.countApproxDistinct(relativeSD)

def name(): String = rdd.name

Expand Down
106 changes: 95 additions & 11 deletions core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.reflect.ClassTag

import com.clearspring.analytics.stream.cardinality.HyperLogLog
import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus
import org.apache.hadoop.conf.{Configurable, Configuration}
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.io.SequenceFile.CompressionType
Expand All @@ -46,7 +46,6 @@ import org.apache.spark.Partitioner.defaultPartitioner
import org.apache.spark.SparkContext._
import org.apache.spark.partial.{BoundedDouble, PartialResult}
import org.apache.spark.serializer.Serializer
import org.apache.spark.util.SerializableHyperLogLog

/**
* Extra functions available on RDDs of (key, value) pairs through an implicit conversion.
Expand Down Expand Up @@ -214,18 +213,94 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
}

/**
* :: Experimental ::
*
* Return approximate number of distinct values for each key in this RDD.
* The accuracy of approximation can be controlled through the relative standard deviation
* (relativeSD) parameter, which also controls the amount of memory used. Lower values result in
* more accurate counts but increase the memory footprint and vice versa. Uses the provided
* Partitioner to partition the output RDD.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*
* @param p The precision value for the normal set.
* <code>p</code> must be a value between 4 and <code>sp</code> (32 max).
* @param sp The precision value for the sparse set, between 0 and 32.
* If <code>sp</code> equals 0, the sparse representation is skipped.
* @param partitioner Partitioner to use for the resulting RDD.
*/
def countApproxDistinctByKey(relativeSD: Double, partitioner: Partitioner): RDD[(K, Long)] = {
val createHLL = (v: V) => new SerializableHyperLogLog(new HyperLogLog(relativeSD)).add(v)
val mergeValueHLL = (hll: SerializableHyperLogLog, v: V) => hll.add(v)
val mergeHLL = (h1: SerializableHyperLogLog, h2: SerializableHyperLogLog) => h1.merge(h2)
@Experimental
def countApproxDistinctByKey(p: Int, sp: Int, partitioner: Partitioner): RDD[(K, Long)] = {
val createHLL = (v: V) => {
val hll = new HyperLogLogPlus(p, sp)
hll.offer(v)
hll
}
val mergeValueHLL = (hll: HyperLogLogPlus, v: V) => {
hll.offer(v)
hll
}
val mergeHLL = (h1: HyperLogLogPlus, h2: HyperLogLogPlus) => {
h1.addAll(h2)
h1
}

combineByKey(createHLL, mergeValueHLL, mergeHLL, partitioner).mapValues(_.cardinality())
}

/**
* :: Experimental ::
*
* Return approximate number of distinct values for each key in this RDD.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*
* @param p The precision value for the normal set.
* <code>p</code> must be a value between 4 and <code>sp</code> (32 max).
* @param sp The precision value for the sparse set, between 0 and 32.
* If <code>sp</code> equals 0, the sparse representation is skipped.
* @param numPartitions Number of partitions in the resulting RDD.
*/
@Experimental
def countApproxDistinctByKey(p: Int, sp: Int, numPartitions: Int): RDD[(K, Long)] = {
countApproxDistinctByKey(p, sp, new HashPartitioner(numPartitions))
}

/**
* :: Experimental ::
*
* Return approximate number of distinct values for each key in this RDD.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*
* @param p The precision value for the normal set.
* <code>p</code> must be a value between 4 and <code>sp</code> (32 max).
* @param sp The precision value for the sparse set, between 0 and 32.
* If <code>sp</code> equals 0, the sparse representation is skipped.
*/
@Experimental
def countApproxDistinctByKey(p: Int, sp: Int): RDD[(K, Long)] = {
countApproxDistinctByKey(p, sp, defaultPartitioner(self))
}

combineByKey(createHLL, mergeValueHLL, mergeHLL, partitioner).mapValues(_.value.cardinality())
/**
* Return approximate number of distinct values for each key in this RDD.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*
* @param relativeSD The relative standard deviation for the counter.
* Smaller values create counters that require more space.
* @param partitioner Partitioner to use for the resulting RDD
*/
@deprecated("Use countApproxDistinctByKey with parameter p and sp", "1.0.1")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shall we put the mapping I mentioned below to the deprecation message?

def countApproxDistinctByKey(relativeSD: Double, partitioner: Partitioner): RDD[(K, Long)] = {
// See stream-lib's HyperLogLog implementation on the conversion from relativeSD to p.
val p = (math.log((1.106 / relativeSD) * (1.106 / relativeSD)) / math.log(2)).toInt
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I checked the paper. I think the correct mapping should be

val p = math.ceil(2.0 * math.log(1.054 / relativeSD) / math.log(2)).toInt

HyperLogLogPlus uses 64-bit hash. The constant beta is 1.054 in the paper. To guarantee the precision, we need ceil. We also need to check whether p is out of bound if relativeSD is really small.

countApproxDistinctByKey(p, 0, partitioner)
}

/**
Expand All @@ -235,7 +310,11 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
* more accurate counts but increase the memory footprint and vice versa. HashPartitions the
* output RDD into numPartitions.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*/
@deprecated("Use countApproxDistinctByKey with parameter p and sp", "1.0.1")
def countApproxDistinctByKey(relativeSD: Double, numPartitions: Int): RDD[(K, Long)] = {
countApproxDistinctByKey(relativeSD, new HashPartitioner(numPartitions))
}
Expand All @@ -247,7 +326,12 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
* more accurate counts but increase the memory footprint and vice versa. The default value of
* relativeSD is 0.05. Hash-partitions the output RDD using the existing partitioner/parallelism
* level.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*/
@deprecated("Use countApproxDistinctByKey with parameter p and sp", "1.0.1")
def countApproxDistinctByKey(relativeSD: Double = 0.05): RDD[(K, Long)] = {
countApproxDistinctByKey(relativeSD, defaultPartitioner(self))
}
Expand Down
45 changes: 37 additions & 8 deletions core/src/main/scala/org/apache/spark/rdd/RDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.reflect.{classTag, ClassTag}

import com.clearspring.analytics.stream.cardinality.HyperLogLog
import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus
import org.apache.hadoop.io.BytesWritable
import org.apache.hadoop.io.compress.CompressionCodec
import org.apache.hadoop.io.NullWritable
Expand All @@ -41,7 +41,7 @@ import org.apache.spark.partial.CountEvaluator
import org.apache.spark.partial.GroupedCountEvaluator
import org.apache.spark.partial.PartialResult
import org.apache.spark.storage.StorageLevel
import org.apache.spark.util.{BoundedPriorityQueue, SerializableHyperLogLog, Utils}
import org.apache.spark.util.{BoundedPriorityQueue, Utils}
import org.apache.spark.util.collection.OpenHashMap
import org.apache.spark.util.random.{BernoulliSampler, PoissonSampler}

Expand Down Expand Up @@ -921,15 +921,44 @@ abstract class RDD[T: ClassTag](
* :: Experimental ::
* Return approximate number of distinct elements in the RDD.
*
* The accuracy of approximation can be controlled through the relative standard deviation
* (relativeSD) parameter, which also controls the amount of memory used. Lower values result in
* more accurate counts but increase the memory footprint and vise versa. The default value of
* relativeSD is 0.05.
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*
* @param p The precision value for the normal set.
* <code>p</code> must be a value between 4 and <code>sp</code>.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let us put the following into the doc.

The relative standard error is approximately 1.054 / sqrt(2^p). Setting a nonzero sp > p would trigger sparse representation of registers, which may reduce the memory consumption and increase accuracy WHEN the cardinality is small.

* @param sp The precision value for the sparse set, between 0 and 32.
* If <code>sp</code> equals 0, the sparse representation is skipped.
*/
@Experimental
def countApproxDistinct(p: Int, sp: Int): Long = {
require(p >= 4, s"p ($p) must be greater than 0")
require(sp <= 32, s"sp ($sp) cannot be greater than 32")
require(sp == 0 || p <= sp, s"p ($p) cannot be greater than sp ($sp)")
val zeroCounter = new HyperLogLogPlus(p, sp)
aggregate(zeroCounter)(
(hll: HyperLogLogPlus, v: T) => {
hll.offer(v)
hll
},
(h1: HyperLogLogPlus, h2: HyperLogLogPlus) => {
h1.addAll(h2)
h2
}).cardinality()
}

/**
* Return approximate number of distinct elements in the RDD.
*
* The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
* Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available at
* [[http://research.google.com/pubs/pub40671.html]].
*/
@deprecated("Use countApproxDistinct with parameter p and sp", "1.0.1")
def countApproxDistinct(relativeSD: Double = 0.05): Long = {
val zeroCounter = new SerializableHyperLogLog(new HyperLogLog(relativeSD))
aggregate(zeroCounter)(_.add(_), _.merge(_)).value.cardinality()
// See stream-lib's HyperLogLog implementation on the conversion from relativeSD to p.
val p = (math.log((1.106 / relativeSD) * (1.106 / relativeSD)) / math.log(2)).toInt
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See the formula I mentioned above.

countApproxDistinct(p, 0)
}

/**
Expand Down
Loading