Skip to content

[SPARK-7826][CORE] Suppress extra calling getCacheLocs. #6352

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
ueshin wants to merge 9 commits into from
Closed

[SPARK-7826][CORE] Suppress extra calling getCacheLocs. #6352

Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
9a80fad
Suppress extra calling getCacheLocs.
ueshin May 22, 2015
a4d944a
Add an unit test.
ueshin May 23, 2015
8248386
Revert "Suppress extra calling getCacheLocs."
ueshin May 23, 2015
f87f2ec
Get cached locations from block manager only if the storage level of …
ueshin May 23, 2015
b9c835c
Put the condition that checks if the RDD has uncached partition or no…
ueshin May 23, 2015
6f3125c
Fix scalastyle.
ueshin May 23, 2015
d858b59
Move the storageLevel check inside the if (!cacheLocs.contains(rdd.id…
ueshin May 24, 2015
10b1b22
Simplify the unit test.
ueshin May 27, 2015
3d4d036
Modify a test and the documentation.
ueshin May 28, 2015
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
10 changes: 6 additions & 4 deletions core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -386,13 +386,15 @@ class DAGScheduler(
def visit(rdd: RDD[_]) {
if (!visited(rdd)) {
visited += rdd
if (getCacheLocs(rdd).contains(Nil)) {
if (rdd.dependencies.size < 2 || getCacheLocs(rdd).contains(Nil)) {

@JoshRosen JoshRosen May 23, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

As a general aside, I find getCacheLocs(rdd).contains(Nil) to be hard to understand to begin with. I think that this condition is meant to be read as "if at least one partition of this RDD is not cached anywhere...". Maybe this code would be easier to review / parse if we extracted this condition into a variable, perhaps a lazy val if we want to short-circuit, named rddHasUncachedPartitions, or !rddIsCached if we don't mind negation.

for (dep <- rdd.dependencies) {
dep match {
case shufDep: ShuffleDependency[_, _, _] =>
val mapStage = getShuffleMapStage(shufDep, stage.jobId)
if (!mapStage.isAvailable) {
missing += mapStage
if (getCacheLocs(rdd).contains(Nil)) {
val mapStage = getShuffleMapStage(shufDep, stage.jobId)
if (!mapStage.isAvailable) {
missing += mapStage
}
}
case narrowDep: NarrowDependency[_] =>
waitingForVisit.push(narrowDep.rdd)
Expand Down