You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by ConeyLiu <gi...@git.apache.org> on 2017/09/22 04:05:35 UTC
[GitHub] spark pull request #19317: [SPARK-22098][CORE] Add new method aggregateByKey...
GitHub user ConeyLiu opened a pull request:
https://github.com/apache/spark/pull/19317
[SPARK-22098][CORE] Add new method aggregateByKeyLocally in RDD
## What changes were proposed in this pull request?
https://issues.apache.org/jira/browse/SPARK-22096
NaiveBayes currently takes aggreateByKey followed by a collect to calculate frequency for each feature/label. We can implement a new function 'aggregateByKeyLocally' in RDD that merges locally on each mapper before sending results to a reducer to save one stage.
We tested on NaiveBayes and see ~20% performance gain with these changes.
This is a subtask of our improvement.
## How was this patch tested?
New UT.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/ConeyLiu/spark aggregatebykeylocally
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/19317.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 #19317
----
commit 73a85dc5963ac46f181a9499deabb18da4ccc308
Author: Xianyang Liu <xi...@intel.com>
Date: 2017-08-31T05:16:09Z
add new method 'aggregateByKeyLocally'
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19317: [SPARK-22098][CORE] Add new method aggregateByKey...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu commented on a diff in the pull request:
https://github.com/apache/spark/pull/19317#discussion_r144760426
--- Diff: core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala ---
@@ -180,6 +180,56 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
* as in scala.TraversableOnce. The former operation is used for merging values within a
* partition, and the latter is used for merging values between partitions. To avoid memory
* allocation, both of these functions are allowed to modify and return their first argument
+ * instead of creating a new U. This method is different from the ordinary "aggregateByKey"
+ * method, it directly returns a map to the driver, rather than a rdd. This will also perform
+ * the merging locally on each mapper before sending results to a reducer, similarly to a
--- End diff --
`aggregateByKey(...).toLocalIterator` need a shuffle for `aggregateByKey` and then collect the `RDD` to driver as a iterator. But `aggregateByKeyLocally` seems like the `aggregateByKey`, while there isn't a shuffle. It calculates the combines in each task and then collect all the `map` direcly to driver and do the finally combines on driver.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu commented on the issue:
https://github.com/apache/spark/pull/19317
Hi @WeichenXu123, any comments on this?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by WeichenXu123 <gi...@git.apache.org>.
Github user WeichenXu123 commented on the issue:
https://github.com/apache/spark/pull/19317
Yes. I guess the perf gain is because, this PR use local hashmap which can use unlimited memory, but current spark aggregation impl, will auto spill local hashmap when exceeding a threshold.
Memory management is a difficult thing we should not try to do it in user code I think, because we're hard to estimate how huge the hashmap will grow to.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19317: [SPARK-22098][CORE] Add new method aggregateByKey...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/19317#discussion_r144753719
--- Diff: core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala ---
@@ -180,6 +180,56 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
* as in scala.TraversableOnce. The former operation is used for merging values within a
* partition, and the latter is used for merging values between partitions. To avoid memory
* allocation, both of these functions are allowed to modify and return their first argument
+ * instead of creating a new U. This method is different from the ordinary "aggregateByKey"
+ * method, it directly returns a map to the driver, rather than a rdd. This will also perform
+ * the merging locally on each mapper before sending results to a reducer, similarly to a
--- End diff --
doesn't `aggregateByKey` perform map side combine?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu commented on the issue:
https://github.com/apache/spark/pull/19317
OK, just keep it. Does this need more test or more improvements ?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu commented on the issue:
https://github.com/apache/spark/pull/19317
cc @VinceShieh
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu commented on the issue:
https://github.com/apache/spark/pull/19317
Does not `treeAggregate` will introduce another `Shuffle`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by WeichenXu123 <gi...@git.apache.org>.
Github user WeichenXu123 commented on the issue:
https://github.com/apache/spark/pull/19317
It is better adding more perf test for `OpenHashSet` replacement to avoid perf regression. And I found `reduceByKeyLocally` also use `JHashSet`, I am not sure whether there is some special reason. ping @cloud-fan Can you help confirm this ? I cannot find the original author for that.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by WeichenXu123 <gi...@git.apache.org>.
Github user WeichenXu123 commented on the issue:
https://github.com/apache/spark/pull/19317
Oh I get your point. This is different from `RDD.aggregate`, it directly return Map and avoid shuffling. it seems useful when numKeys is small.
But, I check the final `reduce` step, it seems can be optimized using `treeAggregate`, and we can add a `depth` parameter.
And using `OpenHashSet` instead of `JHashMap` looks better, but we need test first.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19317: [SPARK-22098][CORE] Add new method aggregateByKey...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/19317#discussion_r144767175
--- Diff: core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala ---
@@ -180,6 +180,56 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
* as in scala.TraversableOnce. The former operation is used for merging values within a
* partition, and the latter is used for merging values between partitions. To avoid memory
* allocation, both of these functions are allowed to modify and return their first argument
+ * instead of creating a new U. This method is different from the ordinary "aggregateByKey"
+ * method, it directly returns a map to the driver, rather than a rdd. This will also perform
+ * the merging locally on each mapper before sending results to a reducer, similarly to a
--- End diff --
BTW `collect all the map directly to driver` may easily OOM the driver, while shuffling to multiple reducers can reduce the memory pressure. Even if only shuffle to one reducer, it still better as the executor usually have more memory than the driver.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by jiangxb1987 <gi...@git.apache.org>.
Github user jiangxb1987 commented on the issue:
https://github.com/apache/spark/pull/19317
cc @WeichenXu123 mind take a look?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu commented on the issue:
https://github.com/apache/spark/pull/19317
@jiangxb1987 ,@WeichenXu123, thanks for your reviewing. This change is inspired by the `TODO List`. You can see the follow code snippet:
```scala
// TODO: Calling aggregateByKey and collect creates two stages, we can implement something
// TODO: similar to reduceByKeyLocally to save one stage.
val aggregated = dataset.select(col($(labelCol)), w, col($(featuresCol))).rdd
.map { row => (row.getDouble(0), (row.getDouble(1), row.getAs[Vector](2)))
}.aggregateByKey[(Double, DenseVector)]((0.0, Vectors.zeros(numFeatures).toDense))(
seqOp = {
case ((weightSum: Double, featureSum: DenseVector), (weight, features)) =>
requireValues(features)
BLAS.axpy(weight, features, featureSum)
(weightSum + weight, featureSum)
},
combOp = {
case ((weightSum1, featureSum1), (weightSum2, featureSum2)) =>
BLAS.axpy(1.0, featureSum2, featureSum1)
(weightSum1 + weightSum2, featureSum1)
}).collect().sortBy(_._1)
```
- The code `aggregateByKeyLocally` we implemented is similar to the `reduceByKeyLocally `.
- I agree with your suggestion for using `OpenHashSet ` instead of `JHashMap`. I could change it, and also the `reduceByKeyLocally` maybe need a change to.
- Because here we collect all the aggregated data to driver and sort it. I think the data could be small. And the collected data of two implements could be almost equally.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by WeichenXu123 <gi...@git.apache.org>.
Github user WeichenXu123 commented on the issue:
https://github.com/apache/spark/pull/19317
@ConeyLiu Yes tree aggregate introduce extra shuffle. But it is possible to improve perf when driver total collecting data size from executors are large and there're many partitions.
But I think we can keep the same with `reduceByKeyLocally` for now. This is possible optimization which can be done in future.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19317: [SPARK-22098][CORE] Add new method aggregateByKey...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu commented on a diff in the pull request:
https://github.com/apache/spark/pull/19317#discussion_r145294297
--- Diff: core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala ---
@@ -180,6 +180,56 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
* as in scala.TraversableOnce. The former operation is used for merging values within a
* partition, and the latter is used for merging values between partitions. To avoid memory
* allocation, both of these functions are allowed to modify and return their first argument
+ * instead of creating a new U. This method is different from the ordinary "aggregateByKey"
+ * method, it directly returns a map to the driver, rather than a rdd. This will also perform
+ * the merging locally on each mapper before sending results to a reducer, similarly to a
--- End diff --
thansk for the advance, I'll close it and try `mapPartitions(...).collect` in `NaiveBayes`.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu commented on the issue:
https://github.com/apache/spark/pull/19317
Test case:
```scala
test("performance of aggregateByKeyLocally ") {
val random = new Random(1)
val pairs = sc.parallelize(0 until 10000000, 20)
.map(p => (random.nextInt(100), p))
.persist(StorageLevel.MEMORY_ONLY)
pairs.count()
val start = System.currentTimeMillis()
// val jHashMap = pairs.aggregateByKeyLocallyWithJHashMap(new HashSet[Int]())(_ += _, _ ++= _).toArray
val openHashMap = pairs.aggregateByKeyLocally(new HashSet[Int]())(_ += _, _ ++= _).toArray
println(System.currentTimeMillis() - start)
}
```
Test result:
| map| 1| 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | avg |
| ------| ------ | ------ |------| ------ | ------ |------| ------ | ------ |------| ------ | ------ |
| JHashMap | 2921 | 2920 | 2843 | 2950 | 2898 | 3316 | 2770 | 2994 | 3016 | 3005 | 2963.3 |
| OpenHashMap | 3029 | 2884 | 3064 | 3023 | 3108 | 3194 | 3003 | 2961 | 3115 | 3023 | 3040.4 |
Looks almost the same performance.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19317: [SPARK-22098][CORE] Add new method aggregateByKey...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/19317#discussion_r144766776
--- Diff: core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala ---
@@ -180,6 +180,56 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
* as in scala.TraversableOnce. The former operation is used for merging values within a
* partition, and the latter is used for merging values between partitions. To avoid memory
* allocation, both of these functions are allowed to modify and return their first argument
+ * instead of creating a new U. This method is different from the ordinary "aggregateByKey"
+ * method, it directly returns a map to the driver, rather than a rdd. This will also perform
+ * the merging locally on each mapper before sending results to a reducer, similarly to a
--- End diff --
> collect all the `map` directly to driver
technically it's a shuffle too, and generally `aggregateByKey` is better for the following reasons:
1. The final combine is executed on multiple reducers, which has better parallelism than doing it on the driver.
2. We should always prefer doing computation on executors instead of the driver, because the driver is responsible for scheduling and has a high cost of failure recovery.
3. using spark shuffle is better for fault tolerant. If one reducer failed, you don't need to rerun all mappers.
So I'm -1 on this API. If there are special cases we wanna to local aggregate, just call `RDD.map.collect` and do the local aggregate manually.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19317: [SPARK-22098][CORE] Add new method aggregateByKey...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu commented on a diff in the pull request:
https://github.com/apache/spark/pull/19317#discussion_r144755666
--- Diff: core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala ---
@@ -180,6 +180,56 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
* as in scala.TraversableOnce. The former operation is used for merging values within a
* partition, and the latter is used for merging values between partitions. To avoid memory
* allocation, both of these functions are allowed to modify and return their first argument
+ * instead of creating a new U. This method is different from the ordinary "aggregateByKey"
+ * method, it directly returns a map to the driver, rather than a rdd. This will also perform
+ * the merging locally on each mapper before sending results to a reducer, similarly to a
--- End diff --
Yeah, it will. Here the 'difference' means it directly returns a map to the driver rather than an RDD.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by VinceShieh <gi...@git.apache.org>.
Github user VinceShieh commented on the issue:
https://github.com/apache/spark/pull/19317
Nice catch. thanks. the perf gain is truly narrow.
I believe this impl just tried to align with the impl of 'reduceByKeyLocally'.
@ConeyLiu maybe we should revisit the code, along with the 'reduceByKeyLocally' impl.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by WeichenXu123 <gi...@git.apache.org>.
Github user WeichenXu123 commented on the issue:
https://github.com/apache/spark/pull/19317
And I have to point out that your impl have high risk causing OOM. The current impl will auto spill when local hashmap is too large and can take advantage of spark auto memory management mechanism which you'd better take a look.
Another thing is the JHashmap will be slow perf and it is better to use `org.apache.spark.util.collection.OpenHashSet`, in the case the hashmap is append-only.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19317: [SPARK-22098][CORE] Add new method aggregateByKey...
Posted by ConeyLiu <gi...@git.apache.org>.
Github user ConeyLiu closed the pull request at:
https://github.com/apache/spark/pull/19317
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #19317: [SPARK-22098][CORE] Add new method aggregateByKey...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/19317#discussion_r144757881
--- Diff: core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala ---
@@ -180,6 +180,56 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
* as in scala.TraversableOnce. The former operation is used for merging values within a
* partition, and the latter is used for merging values between partitions. To avoid memory
* allocation, both of these functions are allowed to modify and return their first argument
+ * instead of creating a new U. This method is different from the ordinary "aggregateByKey"
+ * method, it directly returns a map to the driver, rather than a rdd. This will also perform
+ * the merging locally on each mapper before sending results to a reducer, similarly to a
--- End diff --
what's the difference between `aggregateByKeyLocally` and `aggregateByKey(...).toLocalIterator`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #19317: [SPARK-22098][CORE] Add new method aggregateByKeyLocally...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/19317
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org