Skip to content

[WIP][SPARK-14408][CORE] Changed RDD.treeAggregate to use fold instead of reduce #12217

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

Closed
wants to merge 5 commits into from
Closed

[WIP][SPARK-14408][CORE] Changed RDD.treeAggregate to use fold instead of reduce #12217

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
93e3cb3
Changed RDD.treeAggregate to use fold instead of reduce
jkbradley Apr 6, 2016
0d05b96
Still testing treeAggregate implementations
jkbradley Apr 7, 2016
7fcc10d
fixed bug in treeAgg test
jkbradley Apr 8, 2016
16e79e3
Fixed incorrect statement about failure
jkbradley Apr 8, 2016
02d107a
Fixed bug in treeAggregate using fold
jkbradley Apr 8, 2016
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
9 changes: 5 additions & 4 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 @@ -1061,7 +1061,7 @@ abstract class RDD[T: ClassTag](
* Aggregates the elements of this RDD in a multi-level tree pattern.
*
* @param depth suggested depth of the tree (default: 2)
* @see [[org.apache.spark.rdd.RDD#aggregate]]
* @see [[org.apache.spark.rdd.RDD#aggregate]] These two methods have identical semantics.
Copy link
Member

@HyukjinKwon HyukjinKwon Jun 3, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

Just to help ... I believe the actual Javadoc errors look ...

[error] /home/jenkins/workspace/NewSparkPullRequestBuilder/core/target/java/org/apache/spark/rdd/RDD.java:660: error: unexpected content
[error]    * @see {@link org.apache.spark.rdd.RDD#aggregate} These two methods have identical semantics.
[error]      ^

*/
def treeAggregate[U: ClassTag](zeroValue: U)(
seqOp: (U, T) => U,
Expand All @@ -1075,7 +1075,7 @@ abstract class RDD[T: ClassTag](
val cleanCombOp = context.clean(combOp)
val aggregatePartition =
(it: Iterator[T]) => it.aggregate(zeroValue)(cleanSeqOp, cleanCombOp)
var partiallyAggregated = mapPartitions(it => Iterator(aggregatePartition(it)))
var partiallyAggregated: RDD[U] = mapPartitions(it => Iterator(aggregatePartition(it)))
var numPartitions = partiallyAggregated.partitions.length
val scale = math.max(math.ceil(math.pow(numPartitions, 1.0 / depth)).toInt, 2)
// If creating an extra level doesn't help reduce
Expand All @@ -1087,9 +1087,10 @@ abstract class RDD[T: ClassTag](
val curNumPartitions = numPartitions
partiallyAggregated = partiallyAggregated.mapPartitionsWithIndex {
(i, iter) => iter.map((i % curNumPartitions, _))
}.reduceByKey(new HashPartitioner(curNumPartitions), cleanCombOp).values
}.foldByKey(zeroValue, new HashPartitioner(curNumPartitions))(cleanCombOp).values
}
partiallyAggregated.reduce(cleanCombOp)
val copiedZeroValue = Utils.clone(zeroValue, sc.env.closureSerializer.newInstance())
partiallyAggregated.fold(copiedZeroValue)(cleanCombOp)
Copy link
Member Author

Choose a reason for hiding this comment

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

It's this line which makes AFTSurvivalRegression fail. Not sure why...

Copy link
Member

Choose a reason for hiding this comment

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

@jkbradley Is it because the code uses zeroValue, possibly modifying it, before you copy it? what about copying it before line 1085?

Copy link
Member Author

Choose a reason for hiding this comment

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

I tried making one copy for each use of zeroValue at the beginning of the method, but that didn't fix the AFT test failures.

Copy link
Member

Choose a reason for hiding this comment

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

Hm OK if you've got a copy for each of the 3 usages, that really can't be it. Unless the clone isn't implemented as a deep clone for the object in question somehow. Could it be due to a different order of applying the combOp in this case? that's the only other thing I can think of if this change alone is the issue.

Copy link
Member Author

Choose a reason for hiding this comment

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

The ordering of the combOp really shouldn't matter for AFT. I feel like it must be some esoteric closure issue.

}
}

Expand Down
30 changes: 29 additions & 1 deletion core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,22 @@ class RDDSuite extends SparkFunSuite with SharedSparkContext {
assert(ser.serialize(union.partitions.head).limit() < 2000)
}

test("fold") {
val rdd = sc.makeRDD(-1000 until 1000, 10)
def op: (Int, Int) => Int = (c: Int, x: Int) => c + x
val sum = rdd.fold(0)(op)
assert(sum === -1000)
}

test("fold with op modifying first arg") {
val rdd = sc.makeRDD(-1000 until 1000, 10).map(x => Array(x))
def op: (Array[Int], Array[Int]) => Array[Int] = { (c: Array[Int], x: Array[Int]) =>
c(0) += x(0)
c
}
val sum = rdd.fold(Array(0))(op)
}

test("aggregate") {
val pairs = sc.makeRDD(Array(("a", 1), ("b", 2), ("a", 2), ("c", 5), ("a", 3)))
type StringMap = HashMap[String, Int]
Expand All @@ -185,7 +201,19 @@ class RDDSuite extends SparkFunSuite with SharedSparkContext {
def combOp: (Long, Long) => Long = (c1: Long, c2: Long) => c1 + c2
for (depth <- 1 until 10) {
val sum = rdd.treeAggregate(0L)(seqOp, combOp, depth)
assert(sum === -1000L)
assert(sum === -1000)
}
}

test("treeAggregate with ops modifying first args") {
val rdd = sc.makeRDD(-1000 until 1000, 10).map(x => Array(x))
def op: (Array[Int], Array[Int]) => Array[Int] = { (c: Array[Int], x: Array[Int]) =>
c(0) += x(0)
c
}
for (depth <- 1 until 10) {
val sum = rdd.treeAggregate(Array(0))(op, op, depth)
assert(sum(0) === -1000)
}
}

Expand Down