[SPARK-1021] Defer the data-driven computation of partition bounds in so... #1689

erikerlandson · 2014-07-31T14:27:02Z

...rtByKey() until evaluation.

AmplabJenkins · 2014-07-31T14:27:29Z

Can one of the admins verify this patch?

JoshRosen · 2014-07-31T21:56:09Z

Jenkins, this is ok to test.

SparkQA · 2014-07-31T21:59:06Z

QA tests have started for PR 1689. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17611/consoleFull

SparkQA · 2014-07-31T22:50:37Z

QA results for PR 1689:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17611/consoleFull

rxin · 2014-08-06T20:17:17Z

core/src/main/scala/org/apache/spark/Partitioner.scala

Can we perhaps make this thread safe?

SparkQA · 2014-08-07T01:49:31Z

QA tests have started for PR 1689. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18089/consoleFull

SparkQA · 2014-08-07T02:39:54Z

QA results for PR 1689:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18089/consoleFull

rxin · 2014-08-07T05:47:40Z

core/src/main/scala/org/apache/spark/Partitioner.scala

Do we not want to deserialize valRB if it is not null? Are you worried rangeBounds might be called while the deserialization is happening?

also was assuming readObject might be called in multiple threads. Can that happen?

that's not possible

erikerlandson · 2014-08-15T17:45:02Z

Latest push updates RangePartition sampling job to be async, and updates the async action functions so that they will properly enclose the sampling job induced by calling 'partitions'.

SparkQA · 2014-08-15T17:45:11Z

QA tests have started for PR 1689 at commit 09f0637.

This patch merges cleanly.

markhamstra · 2014-08-15T18:08:24Z

Excellent! I'll try to find some time to review this soon.

SparkQA · 2014-08-15T18:38:23Z

QA tests have finished for PR 1689 at commit 09f0637.

This patch passes unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2014-08-16T16:30:08Z

QA tests have started for PR 1689 at commit f3448e4.

This patch merges cleanly.

SparkQA · 2014-08-16T17:22:51Z

QA tests have finished for PR 1689 at commit f3448e4.

This patch passes unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2014-09-05T23:44:14Z

Can one of the admins verify this patch?

… sortByKey() until evaluation.

…ePartitioner sampling job properly

rxin · 2014-09-12T20:28:12Z

Jenkins, test this please.

SparkQA · 2014-09-12T20:34:17Z

QA tests have started for PR 1689 at commit 50b6da6.

This patch merges cleanly.

rxin · 2014-09-12T20:44:07Z

@erikerlandson thanks for looking at this.

A few questions:

After this pull request, does anything still use SimpleFutureAction?
If I understand this correctly, this could potentially block the single-threaded scheduler from doing anything else while waiting for the rangeBounds to be computed. Any comment on this?
This is not always lazy still right? See a test case

c.parallelize(1 to 1000).map(x => (x, x)).sortByKey().join(sc.parallelize(1 to 10).map(x=>(x,x)))

SparkQA · 2014-09-12T21:41:24Z

QA tests have finished for PR 1689 at commit 50b6da6.

This patch passes unit tests.
This patch merges cleanly.
This patch adds no public classes.

erikerlandson · 2014-09-15T17:37:23Z

Hi @rxin,

SimpleFutureAction is still referred to in submitJob method, but that doesn't appear to be invoked anywhere. I was reluctant to get rid of it, as it's all experimental, and I could envision use cases for it.
I see your point. I don't currently have any clever ideas to avoid that scenario when it happens.
Very interesting -- so this scenario is triggered because defaultPartitioner starts examining input RDD partitioners, which sets off the job when it trips over the data driven partitioning computation from sortByKey.

My impression is that this whack-a-mole with non-laziness stems from a combination of (a) a data-dependent partitioner(s), with (b) methods that refer to input partitioners as part of the construction of new RDDs. It might be possible to thread some design changes around so that references to partitioning are consistently encapsulated in a Future. Functions such as defaultPartitioner would then also have to return a Future, etc. Or, even more generally, somehow encapsulate all RDD initialization in a Future, with the idea that these futures would finally unwind when some Action was invoked.

However it seems (imo) outside the scope of this particular Jira/PR. Maybe we could start another umbrella Jira to track possible solutions along these lines.

Another orthogonal thought -- you can short circuit all this by providing a partitioner instead of forcing it to be computed from data. That's not as sexy, or widely applicable, as some deeper fix to the problem, but users can do it now as a workaround when it's feasible.

erikerlandson · 2014-09-15T17:44:05Z

Or, maybe just look into playing the same game with the cogrouped RDDs that I did with sortByKey. Don't get into invoking defaultPartitioner until somebody asks for the output RDD's partitioner, etc.

rxin · 2014-09-16T19:10:14Z

Yea I don't think we need to fully solve 3 here.

My main concern with these set of changes is 2, since a single badly behaved RDD can potentially block the (unfortunately single threaded) scheduler forever. Let me think about this a little bit and get back to you.

If you have an idea about how to fix that, feel free to suggest them.

erikerlandson · 2014-09-16T20:09:46Z

So far the best idea I have for (2) is to set some kind of time-out on the evaluation. The bound computation uses subsampling that will (when all goes well) cap the computation at constant time(*). If the timeout triggers, some sub-optimal falback for partitioning might be used. Or just fail the entire evaluation.

(*) more accurately, constant number of samples. the time required could depend on various things.

rxin · 2014-09-27T06:06:40Z

Actually I looked at it again. I don't think it would block the scheduler because we compute partitions outside the scheduler thread. This approach looks good to me!

rxin · 2014-09-27T06:14:59Z

@erikerlandson i'm going to merge this first. Maybe we can do the cleanup later.

rxin · 2014-09-27T06:18:01Z

BTW one thing that would be great to add is a test that makes sure we don't block the main dag scheduler thread. The reason I think we don't block is that we call rdd.partitions.length in submitJob:

  /**
   * Submit a job to the job scheduler and get a JobWaiter object back. The JobWaiter object
   * can be used to block until the the job finishes executing or can be used to cancel the job.
   */
  def submitJob[T, U](
      rdd: RDD[T],
      func: (TaskContext, Iterator[T]) => U,
      partitions: Seq[Int],
      callSite: CallSite,
      allowLocal: Boolean,
      resultHandler: (Int, U) => Unit,
      properties: Properties = null): JobWaiter[U] =
  {
    // Check to make sure we are not launching a task on a partition that does not exist.
    val maxPartitions = rdd.partitions.length

markhamstra · 2014-09-27T06:22:21Z

Have either of you thought about how to coordinate this with Josh's work on SPARK-3626? #2482

marmbrus · 2014-09-29T00:51:50Z

Since this PR was merged the correlationoptimizer14 test has been hanging. We might want to consider rolling back. You can reproduce the problem as follows: sbt -Dspark.hive.whitelist=correlationoptimizer14 hive/test

rxin · 2014-09-29T01:33:57Z

I reverted this commit. @erikerlandson mind taking a look at this problem?

erikerlandson · 2014-09-29T02:13:29Z

@rxin @marmbrus I will check it out

erikerlandson · 2014-11-03T16:50:48Z

@marmbrus, FWIW, the correlationoptimizer14 test appears to be working for me. I ran it using: env _RUN_SQL_TESTS=true _SQL_TESTS_ONLY=true ./dev/run-tests > ~/run-tests-1021.txt 2>&1

Not sure why, but runningsbt -Dspark.hive.whitelist=correlationoptimizer14 hive/test was not causing the test to run in my environment.

…apache#1689) We’ve added checkAllStateStoreProviders for Dedisco project to allow checking all state stores which is a debugging feature. One thing we recently discussed with Dedisco project is, the time spent on such check is not counted by reportTimeTaken which can be observed in stream progress later.

rxin reviewed Aug 6, 2014
View reviewed changes

core/src/main/scala/org/apache/spark/Partitioner.scala Outdated

Copy link

Contributor

rxin Aug 6, 2014

Choose a reason for hiding this comment

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

Can we perhaps make this thread safe?

rxin reviewed Aug 7, 2014
View reviewed changes

erikerlandson mentioned this pull request Aug 25, 2014

[WIP][SQL] SPARK-2360: CSV import to SchemaRDDs #1351

Closed

erikerlandson added 7 commits September 6, 2014 07:19

[SPARK-1021] Defer the data-driven computation of partition bounds in…

ac67195

… sortByKey() until evaluation.

[SPARK-1021] modify range bounds variable to be thread safe

7143f97

RangePartition sampling job -> FutureAction

ca8913e

Fix bug in exception passing with ComplexFutureAction[T]

b2b20e8

tweak async actions to use ComplexFutureAction[T] so they handle Rang…

b88b5d4

…ePartitioner sampling job properly

exception mystery fixed by fixing bug in ComplexFutureAction

4e334a9

use standard getIteratorSize in countAsync

50b6da6

erikerlandson force-pushed the spark-1021-pr branch from f3448e4 to 50b6da6 Compare September 6, 2014 14:21

asfgit closed this in 2d972fd Sep 27, 2014

scwf mentioned this pull request Sep 28, 2014

[SPARK-3705][SQL]add case for VoidObjectInspector to cover NullType #2552

Closed

marmbrus mentioned this pull request Sep 28, 2014

[SQL]fix spark sql hive tests time out issue #2566

Closed

erikerlandson mentioned this pull request Nov 3, 2014

[SPARK-1021] Defer the data-driven computation of partition bounds in so... #3079

Closed

[SPARK-1021] Defer the data-driven computation of partition bounds in so... #1689

[SPARK-1021] Defer the data-driven computation of partition bounds in so... #1689

Uh oh!

Conversation

erikerlandson commented Jul 31, 2014

Uh oh!

AmplabJenkins commented Jul 31, 2014

Uh oh!

JoshRosen commented Jul 31, 2014

Uh oh!

SparkQA commented Jul 31, 2014

Uh oh!

SparkQA commented Jul 31, 2014

Uh oh!

rxin Aug 6, 2014

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 7, 2014

Uh oh!

SparkQA commented Aug 7, 2014

Uh oh!

rxin Aug 7, 2014

Choose a reason for hiding this comment

Uh oh!

erikerlandson Aug 7, 2014

Choose a reason for hiding this comment

Uh oh!

rxin Sep 27, 2014

Choose a reason for hiding this comment

Uh oh!

erikerlandson commented Aug 15, 2014

Uh oh!

SparkQA commented Aug 15, 2014

Uh oh!

markhamstra commented Aug 15, 2014

Uh oh!

SparkQA commented Aug 15, 2014

Uh oh!

SparkQA commented Aug 16, 2014

Uh oh!

SparkQA commented Aug 16, 2014

Uh oh!

SparkQA commented Sep 5, 2014

Uh oh!

rxin commented Sep 12, 2014

Uh oh!

SparkQA commented Sep 12, 2014

Uh oh!

rxin commented Sep 12, 2014

Uh oh!

SparkQA commented Sep 12, 2014

Uh oh!

erikerlandson commented Sep 15, 2014

Uh oh!

erikerlandson commented Sep 15, 2014

Uh oh!

rxin commented Sep 16, 2014

Uh oh!

erikerlandson commented Sep 16, 2014

Uh oh!

rxin commented Sep 27, 2014

Uh oh!

rxin commented Sep 27, 2014

Uh oh!

rxin commented Sep 27, 2014

Uh oh!

markhamstra commented Sep 27, 2014

Uh oh!

marmbrus commented Sep 29, 2014

Uh oh!

rxin commented Sep 29, 2014

Uh oh!

erikerlandson commented Sep 29, 2014

Uh oh!

erikerlandson commented Nov 3, 2014

Uh oh!

Reviewers

Assignees

Labels

Projects