[SPARK-2871] [PySpark] add countApproxDistinct() API

[davies]

RDD.countApproxDistinct(relativeSD=0.05):

 :: Experimental ::
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
<a href="http://dx.doi.org/10.1145/2452376.2452456">here</a>.
This support all the types of objects, which is supported by
Pyrolite, nearly all builtin types.
@param relativeSD Relative accuracy. Smaller values create
counters that require more space.
It must be greater than 0.000017.
>>> n = sc.parallelize(range(1000)).map(str).countApproxDistinct()
>>> 950 < n < 1050
True
>>> n = sc.parallelize([i % 20 for i in range(1000)]).countApproxDistinct()
>>> 18 < n < 22
True

[davies]

@mateiz@JoshRosen this one was separated from #1791, please take a look at it. ( @mateiz had reviewed this part in that PR, sorry for the duplicated reviewing).

[rxin]

can u add a test to make sure that if u have 1000 non-distinct elements (i.e. the same element appearing 1000 times), this doesn't return ~ 1000?

Asking because I'm not sure how pyspark interacts with Java - if it is through byte array, then the hashcode could be wrong for byte arrays.

[davies]

Good point, it helped me to find out an invalid test case of tuple, which will be unpickled as []Object in Java, and the hashCode of it is not determined by content, so I changed it into set([]), which should have similar behavior across Python and Java.

[davies]

I had changed it to do hash calculation in Python, so it can support all hashable types in Python.

[SparkQA]

QA tests have started for PR 2142 at commit dfd2a2a.

• This patch merges cleanly.

[SparkQA]

QA tests have started for PR 2142 at commit 4cba98f.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2142 at commit 4cba98f.

• This patch fails unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

QA tests have finished for PR 2142 at commit dfd2a2a.

• This patch passes unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

QA tests have started for PR 2142 at commit ded624f.

• This patch merges cleanly.

[SparkQA]

QA tests have started for PR 2142 at commit d306492.

• This patch merges cleanly.

[SparkQA]

QA tests have started for PR 2142 at commit 9d2565f.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2142 at commit ded624f.

• This patch fails unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

QA tests have finished for PR 2142 at commit d306492.

• This patch fails unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

QA tests have finished for PR 2142 at commit 9d2565f.

• This patch fails unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

QA tests have started for PR 2142 at commit 2ab157c.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2142 at commit 2ab157c.

• This patch fails unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

QA tests have started for PR 2142 at commit c38c4e4.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2142 at commit c38c4e4.

• This patch passes unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[JoshRosen]

Why this change?

[davies]

In 64-bit machines, the hash of tuple() should be 64 bits (sys.maxint is 64bits).

[JoshRosen]

To check my understanding of the error-correction code:

• Due to hash collisions, we may underestimate the true number of distinct items.
• Given a set of k random 32-bit hashes, the exact probability of at least one collision (from your link) is 1 - e^(-k(k-1)/2*2^32). We can approximate 1 - e^X by X for small X. Therefore, the approximate probability of some hash collision is -k(k-1)/2*2^32, which is roughly k^2/2^33.

I'm a bit confused about how the current correction term works:

c=-sys.maxint*log(1-float(c) /sys.maxint)

It looks like this is correcting for overestimates of the number of distinct elements by subtracting a term based on the collision probability. In general, won't collisions cause us underestimate instead of overestimating?

Maybe we should approach this by treating the true number of distinct items (k) as a random variable and figuring out the maximum likelihood estimator of k given an observation c of the result of countApproxDistinct.

Before we consider that, though, I wonder whether we even need a correction term. Doesn't the Java implementation of countApproxDistinct already introduce hashing errors that are corrected for in its implementation? I don't think that the two levels of hashing will introduce more error, since I think the hashcode of a Java integer should just be its value.

Note: I'm not a statistician; please correct me if I've gotten anything wrong.

[davies]

After discussion with @JoshRosen offline, we realized that it does not need correction in Python if they have the same hash space both in Python and Java, so I changed the has space (mapping to 2^32) and remove the correction.

Jenkins, test this please.

[SparkQA]

QA tests have started for PR 2142 at commit e20da47.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2142 at commit e20da47.

• This patch passes unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[JoshRosen]

Yeah, I don't think we any special correction because Java will use the hashcodes chosen in Python (since Integer.hashcode is just the integer's value).

[JoshRosen]

LGTM, so I've merged this into master. Thanks!