GitHub user maropu opened a pull request:
https://github.com/apache/incubator-hivemall/pull/33
[HIVEMALL-44][SAPRK] Implement a prototype of Join with TopK for
DataFrame/Spark
## What changes were proposed in this pull request?
This pr implemented a prototype for Top-K joins. In Hivemall, `each_top_k`
is useful for practical use cases. On the other hand, there are some cases we
need to join tables then compute Top-K entries.You know we can compute this
query by using regular joins + `each_top_k`. However, we have space to improve
this query more; that is, we compute Top-K entries while processing joins. This
optimization avoids a substantial amount of I/O for joins.
An example query is as follows;
```
val inputDf = Seq(
("user1", 1, 0.3, 0.5),
("user2", 2, 0.1, 0.1),
("user3", 3, 0.8, 0.0),
("user4", 1, 0.9, 0.9),
("user5", 3, 0.7, 0.2),
("user6", 1, 0.5, 0.4),
("user7", 2, 0.6, 0.8)
).toDF("userId", "group", "x", "y")
val masterDf = Seq(
(1, "pos1-1", 0.5, 0.1),
(1, "pos1-2", 0.0, 0.0),
(1, "pos1-3", 0.3, 0.3),
(2, "pos2-3", 0.1, 0.3),
(2, "pos2-3", 0.8, 0.8),
(3, "pos3-1", 0.1, 0.7),
(3, "pos3-1", 0.7, 0.1),
(3, "pos3-1", 0.9, 0.0),
(3, "pos3-1", 0.1, 0.3)
).toDF("group", "position", "x", "y")
// Compute top-1 rows for each group
val distance = sqrt(
pow(inputDf("x") - masterDf("x"), lit(2.0)) +
pow(inputDf("y") - masterDf("y"), lit(2.0))
)
val top1Df = inputDf.join_top_k(
lit(1), masterDf, inputDf("group") === masterDf("group"),
distance.as("score")
)
```
A quick benchmark is as follows;
```
/**
* Java HotSpot(TM) 64-Bit Server VM 1.8.0_31-b13 on Mac OS X 10.10.2
* Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz
*
* top-k join (k=3): Best/Avg Time(ms) Rate(M/s) Per Row(ns)
Relative
*
-------------------------------------------------------------------------------
* join + rank 65959 / 71324 0.0 503223.9
1.0X
* join + each_top_k 66093 / 78864 0.0 504247.3
1.0X
* join_top_k 5013 / 5431 0.0 38249.3
13.2X
*/
```
The APIs this pr added is unstable and we might change them in upcoming
activities.
## What type of PR is it?
Improvement
### What is the Jira issue?
https://issues.apache.org/jira/browse/HIVEMALL-44
## How was this patch tested?
Added tests in `HivemallOpsSuite`
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/maropu/incubator-hivemall HIVEMALL-44
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-hivemall/pull/33.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #33
----
commit a7c70c962a716ec200871c4f4c52d9a61aae2018
Author: Takeshi YAMAMURO <[email protected]>
Date: 2017-01-30T23:57:19Z
Implement a prototype of Join with TopK
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
