Skip to content

[SPARK-20638][Core]Optimize the CartesianRDD to reduce repeatedly data fetching #17898

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 1 commit into from
Closed

[SPARK-20638][Core]Optimize the CartesianRDD to reduce repeatedly data fetching #17898

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
6 changes: 4 additions & 2 deletions core/src/main/scala/org/apache/spark/rdd/CartesianRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,8 +72,10 @@ class CartesianRDD[T: ClassTag, U: ClassTag](

override def compute(split: Partition, context: TaskContext): Iterator[(T, U)] = {
val currSplit = split.asInstanceOf[CartesianPartition]
for (x <- rdd1.iterator(currSplit.s1, context);
y <- rdd2.iterator(currSplit.s2, context)) yield (x, y)
val groupSize = 500;
for (x <- rdd1.iterator(currSplit.s1, context).grouped(groupSize);
y <- rdd2.iterator(currSplit.s2, context).grouped(groupSize);
Copy link
Member

@viirya viirya May 8, 2017

Choose a reason for hiding this comment

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

One disadvantage I can think now is, longer waiting time for first element.

Copy link
Author

@jtengyp jtengyp May 8, 2017

Choose a reason for hiding this comment

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

This is indeed a disadvantage.

Copy link
Member

@srowen srowen May 8, 2017

Choose a reason for hiding this comment

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

Pardon, doesn't this change the type of the result? you're iterating over groupings not elements, and emitting pairs of groups. As in below, but maybe I'm missing something.

scala> val foo = List(1,2,3)
foo: List[Int] = List(1, 2, 3)

scala> val bar = List(4,5,6)
bar: List[Int] = List(4, 5, 6)

scala> for (x <- foo; y <- bar) yield (x, y)
res0: List[(Int, Int)] = List((1,4), (1,5), (1,6), (2,4), (2,5), (2,6), (3,4), (3,5), (3,6))

scala> (for (x <- foo.grouped(2); y <- bar.grouped(2)) yield (x, y)).foreach(println)
(List(1, 2),List(4, 5))
(List(1, 2),List(6))
(List(3),List(4, 5))
(List(3),List(6))

Copy link
Contributor

@mridulm mridulm May 8, 2017

Choose a reason for hiding this comment

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

The actual yield is on (i, j) and not (x, y) - the next line adds the iteration over the groupings :-)

Copy link
Contributor

@mridulm mridulm May 8, 2017

Choose a reason for hiding this comment

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

I agree with @viirya - there is also an implicit assumption of size here : the batch will get deserialized into memory.
By default, we have kept the iterator model going in spark without needing to buffer (iirc).

Copy link
Contributor

@ConeyLiu ConeyLiu May 8, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

I working on this too. But the optimize method maybe similar to the pr which @viirya opened before, cache the second iterator into local. The code is ready, maybe open a pr in recently. In this patch, I worry about whether we can accurately control the size of the buffer. If we should cache it by BlockManager or MemoryConsumer?

Copy link
Member

@srowen srowen May 8, 2017

Choose a reason for hiding this comment

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

Oh haha right. Hm, but isn't this better solved 'upstream' by buffering an iterator somewhere? or just buffering the iterator right here?

i <- x; j <- y) yield (i, j)
}

override def getDependencies: Seq[Dependency[_]] = List(
Expand Down