[SPARK-21039] [Spark Core] Use treeAggregate instead of aggregate in DataFrame.stat.bloomFilter

[rishabhbhardwaj]

What changes were proposed in this pull request?

To use treeAggregate instead of aggregate in DataFrame.stat.bloomFilter to parallelize the operation of merging the bloom filters
(Please fill in changes proposed in this fix)

How was this patch tested?

unit tests passed
(Please explain how this patch was tested. E.g. unit tests, integration tests, manual tests)
(If this patch involves UI changes, please attach a screenshot; otherwise, remove this)

Please review http://spark.apache.org/contributing.html before opening a pull request.

[HyukjinKwon]

I think we need some figures here. Would you test before/after and share some figures?

[lovasoa]

I ran some tests on a cluster with 5 nodes, 5 executors, and 5 threads per executor.
I created a bloomfilter with the parameters : numElements=150M, fpp=10%. This is a bloom filter of around 90MB.
I created the bloom filter on the column o_orderkey of the table ORDERS of TPC-H with a scale factor of 100. The data is stored in a parquet file on HDFS, with 48 partitions.

The cluster

Results

Using RDD.aggregate

stages

tasks

Notice the huge scheduler delay. The executors spend all their time sending their results back to the driver.

total time: 52 seconds

Using RDD.treeAggregate

stages

tasks

first stage

second stage

total time: 17 seconds

[lovasoa]

Smaller data

You might be wondering: what about smaller data ?
Adding a stage to the computation of course adds some overhead.

I repeated the same experiment as mentioned above, but with a scale factor of 1, (and 1.5M elements in the bloom filter).

With RDD.aggregate

With RDD.treeAggregate

conclusion

There is indeed an overhead, but it is quite small. For the creation of a small bloom filter, the old method (with aggregate) takes 0.3 seconds, and the new one (with treeAggregate) 0.6 seconds.

[srowen]

I'm inclined to use treeAggregate where possible. I think the win for larger data sets is worthwhile.

[HyukjinKwon]

LGTM too given both are semantically identical and the same reason with ^.

[SparkQA]

Test build #3796 has finished for PR 18263 at commit 61bb509.

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

[srowen]

Merged to master