You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/10/12 23:36:43 UTC
[GitHub] [spark] JoshRosen opened a new pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
JoshRosen opened a new pull request #34265:
URL: https://github.com/apache/spark/pull/34265
### What changes were proposed in this pull request?
This PR fixes a longstanding issue where the `DAGScheduler'`s single-threaded event processing loop could become blocked by slow `RDD.getPartitions()` calls, preventing other events (like task completions and concurrent job submissions) from being processed in a timely manner.
With this patch's change, Spark will now call `.partitions` on every RDD in the DAG before submitting a job to the scheduler, ensuring that the expensive `getPartitions()` calls occur outside of the scheduler event loop.
#### Background
The `RDD.partitions` method lazily computes an RDD's partitions by calling `RDD.getPartitions()`. The `getPartitions()` method is invoked only once per RDD and its result is cached in the `RDD.partitions_` private field. Sometimes the `getPartitions()` call can be expensive: for example, `HadoopRDD.getPartitions()` performs file listing operations.
The `.partitions` method is invoked at many different places in Spark's code, including many existing call sites that are outside of the scheduler event loop. As a result, it's _often_ the case that an RDD's partitions will have been computed before the RDD is submitted to the DAGScheduler. For example, [`submitJob` calls `rdd.partitions.length`](https://github.com/apache/spark/blob/3ba57f5edc5594ee676249cd309b8f0d8248462e/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L837), so the DAG root's partitions will be computed outside of the scheduler event loop.
However, there's still some cases where `partitions` gets evaluated for the first time inside of the `DAGScheduler` internals. For example, [`ShuffledRDD.getPartitions`](https://github.com/apache/spark/blob/3ba57f5edc5594ee676249cd309b8f0d8248462e/core/src/main/scala/org/apache/spark/rdd/ShuffledRDD.scala#L92-L94) doesn't call `.partitions` on the RDD being shuffled, so a plan with a ShuffledRDD at the root won't necessarily result in `.partitions` having been called on all RDDs prior to scheduler job submission.
#### Correctness: proving that we make no excess `.partitions` calls
This PR adds code to traverse the DAG prior to job submission and call `.partitions` on every RDD encountered.
I'd like to argue that this results in no _excess_ `.partitions` calls: in every case where the new code calls `.partitions` there is existing code which would have called `.partitions` at some point during a successful job execution:
- Assume that this is the first time we are computing every RDD in the DAG.
- Every RDD appears in some stage.
- [`submitStage` will call `submitMissingTasks`](https://github.com/databricks/runtime/blob/1e83dfe4f685bad7f260621e77282b1b4cf9bca4/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1438) on every stage root RDD.
- [`submitStage` calls `getPreferredLocsInternal`](https://github.com/databricks/runtime/blob/1e83dfe4f685bad7f260621e77282b1b4cf9bca4/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1687-L1696) on every stage root RDD.
- [`getPreferredLocsInternal`](https://github.com/databricks/runtime/blob/1e83dfe4f685bad7f260621e77282b1b4cf9bca4/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2995-L3043) visits the RDD and all of its parents RDDs that are computed in the same stage (via narrow dependencies) and calls `.partitions` on each RDD visited.
- Therefore `.partitions` is invoked on every RDD in the DAG by the time the job has successfully completed.
- Therefore this patch's change does not introduce any new calls to `.partitions` which would not have otherwise occurred (assuming the job succeeded).
#### Ordering of `.partitions` calls
I don't think the order in which `.partitions` calls occur matters for correctness: the DAGScheduler happens to invoke `.partitions` in a particular order today (defined by the DAG traversal order in internal scheduler methods), but there's many lots of out-of-order `.partition` calls occurring elsewhere in the codebase.
#### Handling of exceptions in `.partitions`
I've chosen **not** to add special error-handling for the new `.partitions` calls: if exceptions occur then they'll bubble up, unwrapped, to the user code submitting the Spark job.
It's sometimes important to preserve exception wrapping behavior, but I don't think that concern is warranted in this particular case: whether `getPartitions` occurred inside or outside of the scheduler (impacting whether exceptions manifest in wrapped or unwrapped form, and impacting whether failed jobs appear in the Spark UI) was not crisply defined (and in some rare cases could even be [influenced by Spark settings in non-obvious ways](https://github.com/apache/spark/blob/10d5303174bf4a47508f6227bbdb1eaf4c92fcdb/core/src/main/scala/org/apache/spark/Partitioner.scala#L75-L79)), so I think it's both unlikely that users were relying on the old behavior and very difficult to preserve it.
### Why are the changes needed?
This fixes a longstanding scheduler performance problem which has been reported by multiple users.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
I added a regression test in `BasicSchedulerIntegrationSuite` to cover the regular job submission codepath (`DAGScheduler.submitJob`)This test uses CountDownLatches to simulate the submission of a job containing an RDD with a slow `getPartitions()` call and checks that a concurrently-submitted job is not blocked.
I have **not** added separate integration tests for the `runApproximateJob` and `submitMapStage` codepaths (both of which also received the same fix).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942715545
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/48690/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728609435
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
So, to confirm, do you want to do log info when slow since it only logs once? (I think I didn't get your feedback on this)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728549640
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
Review comment:
Is it equivalent to call `.partitions` on `ShuffleDependency.rdd` only?
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
I wonder if we could switch to log info instead when `.partitions` is suspected to be slow (e.g., when the time > Xs).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen edited a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen edited a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941755354
/cc @squito @dongjoon-hyun @vanzin @srowen @yuchenhuo @jiangxb1987 for review (since they were also tagged in previous PRs for this issue).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm edited a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
mridulm edited a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942664279
> Does this change have any impact on job cancellation? For example, after this change, if I cancel a job shortly after the job is submitted, will I fail to actually cancel this job because the DAGScheduler is still working on resolve the rdd partitions (so it haven't updated the jobIdToStageIds map).
This is happening within submission itself - so caller does not have `jobId` to cancel on (until this completes).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942675025
@jiangxb1987, I don't think this will adversely impact job cancellation via `cancelJobGroup`:
The three `DAGScheduler` methods that I modified are invoked from the main Spark application thread, not the scheduler event thread. At this point no jobs have been submitted.
Just for sake of argument / completeness, let's say that we had a slow `getPartitions` which was executing inside of the scheduler: before, the slow `getPartitions` wouldn't have been cancellable or interruptible at all, since we'd have to wait for it to complete before the job group cancellation event could be processed by the DAG scheduler's event loop.
After this patch's change, that slow work is performed outside of the DAGScheduler. Even though there's no job to cancel, what happens if we're running in a notebook or REPL environment and want to cancel the running cell / command? This depends on the behavior of the notebook/REPL: if the notebook/REPL sends a `Thread.interrupt()` to the running driver thread then then `getPartitions` call might be able exit early depending on whether it's running code which checks for thread interrupts (such as IO, e.g. due to Hadoop filesystem listing operations).
Given this, I think we're okay: we haven't impacted job group cancellation and the driver thread interruption situation is the same as it would be for other slow driver-side operations (such as query planning in Spark SQL).
---
As an aside, I do think it's a bit confusing how the DAGScheduler class has both public methods (called outside the event loop) and private methods used inside the loop mixed into the same class but with different rules around accessing private state. If we wanted to re-architect things then I think it would be clearer to separate those responsibilities into separate classes (maybe something like `DAGSchedulerClient` and `DAGSchedulerBackend`) to more strongly enforce this separation and to prevent accidental access of event-loop state from outside of the loop.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941875238
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/144167/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942682215
> Do you want to add timing info in debug mode ?
@mridulm, good idea: I added logging in 64050f20cecf9f44f61d2f0d2a486ed3153ea329
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941863890
**[Test build #144167 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/144167/testReport)** for PR 34265 at commit [`3bcc554`](https://github.com/apache/spark/commit/3bcc554dfa3dcea8e53ee17c19c09890ec6dcabd).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941809998
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/48645/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941783327
**[Test build #144167 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/144167/testReport)** for PR 34265 at commit [`3bcc554`](https://github.com/apache/spark/commit/3bcc554dfa3dcea8e53ee17c19c09890ec6dcabd).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942687461
**[Test build #144211 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/144211/testReport)** for PR 34265 at commit [`64050f2`](https://github.com/apache/spark/commit/64050f20cecf9f44f61d2f0d2a486ed3153ea329).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941755354
/cc @squito @dongjoon-hyun @vanzin @srowen @yuchenhuo @jiangxb1987 for review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941783327
**[Test build #144167 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/144167/testReport)** for PR 34265 at commit [`3bcc554`](https://github.com/apache/spark/commit/3bcc554dfa3dcea8e53ee17c19c09890ec6dcabd).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728615460
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
To clarify, I think we should tackle user-facing logging in a separate PR, since it involves some design decisions that I'd like to consider separately from this fix.
I think the goal of INFO logging would be to help users explain unexpected pauses in their driver code. A slow `.partitions` call is one source of pauses, but those `.partitions` calls often occur before `eagerlyComputePartitionsForRddAndAncestors` is called. As a result, I think that logging should be added [in RDD.partitions itself](https://github.com/apache/spark/blob/d9b4cc65b89d8497dc4e315395f2c98cc4ac9327/core/src/main/scala/org/apache/spark/rdd/RDD.scala#L292) and not here in `eagerlyComputePartitionsForRddAndAncestors`.
Once we've done that, I'm not sure if there's additional value in promoting the `eagerlyComputePartitionsForRddAndAncestors` logging to INFO: I think that logging is primarily useful for Spark's own developers and not for users.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941847022
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/48645/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728557204
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
Review comment:
I don't think so:
Per the "Correctness: proving that we make no excess .partitions calls" in the PR description, I believe that the `DAGScheduler` will eventually call `.partitions` on every RDD in the DAG.
If we only call `.partitions` on a subset of the RDDs encountered during our DAG traversal here then we run the risk that there could be an RRD whose partitions haven't been evaluated before the `DAGScheduler` calls `.partitions`, potentially leaving us vulnerable to the same performance problem.
It _is_ true that many implementations of `getPartitions()` call `.partitions` on their parent RDDs, but there's no contract which _guarantees_ that in all cases. I think it's fine if we call `.partitions` on something which would have already been computed, though: it'll just [return the stored value](https://github.com/apache/spark/blob/5ac76d9cb45d58eeb4253d50e90060a68c3e87cb/core/src/main/scala/org/apache/spark/rdd/RDD.scala#L289-L300).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941751848
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jiangxb1987 commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
jiangxb1987 commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941854197
Does this change have any impact on job cancellation? For example, after this change, if I cancel a job shortly after the job is submitted, will I fail to actually cancel this job because the DAGScheduler is still working on resolve the rdd partitions (so it haven't updated the `jobIdToStageIds` map).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728599684
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -962,6 +1002,11 @@ private[spark] class DAGScheduler(
throw SparkCoreErrors.cannotRunSubmitMapStageOnZeroPartitionRDDError()
}
+ // SPARK-23626: `RDD.getPartitions()` can be slow, so we eagerly compute
+ // `.partitions` on every RDD in the DAG to ensure that `getPartitions()`
+ // is evaluated outside of the DAGScheduler's single-threaded event loop:
+ eagerlyComputePartitionsForRddAndAncestors(rdd)
+
Review comment:
That's true, but I chose not to make that optimization because (a) it doesn't actually matter from a performance perspective (accessing already-computed `.partitions` is very cheap) and (b) I think the optimization would make the code more complex.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941847022
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/48645/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942743303
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/48690/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728615460
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
To clarify, I think we should tackle user-facing logging in a separate PR, since it involves some design decisions that I'd like to consider separately from this fix.
I think the goal of INFO logging would be to help users explain unexpected pauses in their driver code. A slow `.partitions` call is one source of pauses, but those `.partitions` calls often occur before `eagerlyComputePartitionsForRddAndAncestors` is called. As a result, I think that user-facing warning logging should be added [in RDD.partitions itself](https://github.com/apache/spark/blob/d9b4cc65b89d8497dc4e315395f2c98cc4ac9327/core/src/main/scala/org/apache/spark/rdd/RDD.scala#L292) and not here in `eagerlyComputePartitionsForRddAndAncestors`.
Once we've done that, I'm not sure if there's additional value in promoting the `eagerlyComputePartitionsForRddAndAncestors` logging to INFO: I think that logging is primarily useful for Spark's own developers and not for users.
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
To clarify, I think we should tackle user-facing logging in a separate PR, since it involves some design decisions that I'd like to consider separately from this fix.
I think the goal of INFO logging would be to help users explain unexpected pauses in their driver code. A slow `.partitions` call is one source of pauses, but those `.partitions` calls often occur before `eagerlyComputePartitionsForRddAndAncestors` is called. As a result, I think that user-facing info/warning logging should be added [in RDD.partitions itself](https://github.com/apache/spark/blob/d9b4cc65b89d8497dc4e315395f2c98cc4ac9327/core/src/main/scala/org/apache/spark/rdd/RDD.scala#L292) and not here in `eagerlyComputePartitionsForRddAndAncestors`.
Once we've done that, I'm not sure if there's additional value in promoting the `eagerlyComputePartitionsForRddAndAncestors` logging to INFO: I think that logging is primarily useful for Spark's own developers and not for users.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728566091
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
Review comment:
Ok, just thought if we could avoid some unnecessary calls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941839662
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/48645/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942788084
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/144211/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942760427
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/48690/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728559174
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
If we add logging for slow individual `.partitions` calls then I think we should put that it in `RDD.partitions` itself, not here, since here's many other points in Spark's code where `getPartitions()` might be computed (including at RDD construction time).
What do you think about doing that in a separate followup?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728561177
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
Ah, I see how this might be confusing:
This log line is at the end of `eagerlyComputePartitionsForRddAndAncestors`, not inside of its inner `visit` method, so it's logged only once per call. The RDD id logged here is the ID of the root of the DAG being submitted to the scheduler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728560126
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
I agree that user-facing INFO/WARN messages for slow `getPartitions()` calls might be useful for attributing long gaps in the driver logs.
I also wonder whether users might be better served by placing that logging closer to the actual source of the slowness: for example, `getPartitions()` is most likely going to be slow due to file listing in `HadoopRDD`, so if we place logging in `HadoopRDD` then we can add more information (e.g. "file listing took x seconds and listed y files").
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728616878
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
Thanks for the explanation. Make sense to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941751848
This is a longstanding issue and there's been multiple previous attempts to fix it.
- #3794
- #20770
- #24438
- #27234
Some early attempts were rejected due to thread-safety issues with their approaches or became stale without review.
This PR's approach is very similar to @ajithme's approach in #27234, with a few key differences:
- I allowed exceptions to bubble instead of logging and ignoring them.
- I used a faster and less-race-condition-prone testing approach (using the `SchedulerIntegrationSuite` framework).
- I used a non-recursive tree-traversal method (based on similar existing methods) to avoid stack overflow errors when traversing huge DAGs.
- I also added the fix to `submitMapStage` and `runApproximateJob`: these are much lesser used codepaths but can still potentially benefit from the fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941847022
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942687461
**[Test build #144211 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/144211/testReport)** for PR 34265 at commit [`64050f2`](https://github.com/apache/spark/commit/64050f20cecf9f44f61d2f0d2a486ed3153ea329).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942760427
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/48690/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942788084
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/144211/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942777253
**[Test build #144211 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/144211/testReport)** for PR 34265 at commit [`64050f2`](https://github.com/apache/spark/commit/64050f20cecf9f44f61d2f0d2a486ed3153ea329).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728597530
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -962,6 +1002,11 @@ private[spark] class DAGScheduler(
throw SparkCoreErrors.cannotRunSubmitMapStageOnZeroPartitionRDDError()
}
+ // SPARK-23626: `RDD.getPartitions()` can be slow, so we eagerly compute
+ // `.partitions` on every RDD in the DAG to ensure that `getPartitions()`
+ // is evaluated outside of the DAGScheduler's single-threaded event loop:
+ eagerlyComputePartitionsForRddAndAncestors(rdd)
+
Review comment:
Looks like before the three places calling `eagerlyComputePartitionsForRddAndAncestors`, there are some checks for `rdd.partitions`, so actually we just (need) compute its ancestors?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941847022
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942714763
Thank you for pinging me, @JoshRosen .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941783327
**[Test build #144167 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/144167/testReport)** for PR 34265 at commit [`3bcc554`](https://github.com/apache/spark/commit/3bcc554dfa3dcea8e53ee17c19c09890ec6dcabd).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941875238
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/144167/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941783327
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728560143
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -732,6 +732,35 @@ private[spark] class DAGScheduler(
missing.toList
}
+ /** Invoke `.partitions` on the given RDD and all of its ancestors */
+ private def eagerlyComputePartitionsForRddAndAncestors(rdd: RDD[_]): Unit = {
+ val startTime = System.nanoTime
+ val visitedRdds = new HashSet[RDD[_]]
+ // We are manually maintaining a stack here to prevent StackOverflowError
+ // caused by recursively visiting
+ val waitingForVisit = new ListBuffer[RDD[_]]
+ waitingForVisit += rdd
+
+ def visit(rdd: RDD[_]): Unit = {
+ if (!visitedRdds(rdd)) {
+ visitedRdds += rdd
+
+ // Eagerly compute:
+ rdd.partitions
+
+ for (dep <- rdd.dependencies) {
+ waitingForVisit.prepend(dep.rdd)
+ }
+ }
+ }
+
+ while (waitingForVisit.nonEmpty) {
+ visit(waitingForVisit.remove(0))
+ }
+ logDebug("eagerlyComputePartitionsForRddAndAncestors for RDD %d took %f seconds"
Review comment:
Sorry, I mean the total time rather than the individual `. partitions` calls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yuchenhuo commented on a change in pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
yuchenhuo commented on a change in pull request #34265:
URL: https://github.com/apache/spark/pull/34265#discussion_r728805893
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -841,6 +870,11 @@ private[spark] class DAGScheduler(
"Total number of partitions: " + maxPartitions)
}
+ // SPARK-23626: `RDD.getPartitions()` can be slow, so we eagerly compute
+ // `.partitions` on every RDD in the DAG to ensure that `getPartitions()`
+ // is evaluated outside of the DAGScheduler's single-threaded event loop:
+ eagerlyComputePartitionsForRddAndAncestors(rdd)
Review comment:
Would it be a good idea to add an assertion in the `DebugFilesystem` we have to check that it's not accessed within the event loop thread? It might help catch other cases where event loop might be doing heavy blocking operation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen edited a comment on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
JoshRosen edited a comment on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941755354
/cc @squito @dongjoon-hyun @vanzin @srowen @yuchenhuo @jiangxb1987 for review (since they were also tagged in previous PRs for this issue).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
mridulm commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-942664279
> Does this change have any impact on job cancellation? For example, after this change, if I cancel a job shortly after the job is submitted, will I fail to actually cancel this job because the DAGScheduler is still working on resolve the rdd partitions (so it haven't updated the jobIdToStageIds map).
This is happening within submission itself - so caller does not have `jobId` to cancel on (until this completes and returns).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jiangxb1987 commented on pull request #34265: [SPARK-23626][CORE] Eagerly compute RDD.partitions on entire DAG when submitting job to DAGScheduler
Posted by GitBox <gi...@apache.org>.
jiangxb1987 commented on pull request #34265:
URL: https://github.com/apache/spark/pull/34265#issuecomment-941854197
Does this change have any impact on job cancellation? For example, after this change, if I cancel a job shortly after the job is submitted, will I fail to actually cancel this job because the DAGScheduler is still working on resolve the rdd partitions (so it haven't updated the `jobIdToStageIds` map).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org