Skip to content

[SPARK-27164][Core] RDD.countApprox on empty RDDs schedules jobs which never complete #24100

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
ajithme wants to merge 6 commits into from
Closed

[SPARK-27164][Core] RDD.countApprox on empty RDDs schedules jobs which never complete #24100

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
89caeb0
Fire Jobend event when Result stage has no partition to calculate
ajithme Mar 15, 2019
4881dfd
updated with testcase
ajithme Mar 15, 2019
0920e17
update
ajithme Mar 15, 2019
faaf9e2
update as per review comments
ajithme Mar 16, 2019
bf5bb95
refactor
ajithme Mar 16, 2019
ad7b888
update
ajithme Mar 16, 2019
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
3 changes: 3 additions & 0 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 @@ -1239,6 +1239,9 @@ private[spark] class DAGScheduler(
markMapStageJobsAsFinished(stage)
case stage : ResultStage =>
logDebug(s"Stage ${stage} is actually done; (partitions: ${stage.numPartitions})")

@cloud-fan cloud-fan Mar 15, 2019

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.

would this happen for normal jobs?

@ajithme ajithme Mar 15, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

This condition is guarded by a check when ResultStage itself has zero tasks to run. So it would be skipped for normal jobs

@cloud-fan cloud-fan Mar 15, 2019

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.

I mean, does RDD#count have the same bug when rdd is empty? I think it doesn't but I'm trying to understand why it's an issue only for approximate jobs.

@squito squito Mar 15, 2019

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.

I think this is related to SPARK-26714 / #23637 -- maybe the same change from DAGScheduler.submitJob is needed in DAGScheduler.runApproximateJob

@ajithme ajithme Mar 16, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@cloud-fan, RDD#count scenario is handled via submitJob as initially we have a partitions.size==0 check, just like @squito mentioned

@squito so do you suggest i provide a similar check in runApproximateJob.?? shall i change this PR.?

@ajithme ajithme Mar 16, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I think @squito suggestion is better, i have changed my PR as per it. Please review

val jobId = stage.activeJob.get.jobId
cleanupStateForJobAndIndependentStages(stage.activeJob.get)
listenerBus.post(SparkListenerJobEnd(jobId, clock.getTimeMillis(), JobSucceeded))
}
submitWaitingChildStages(stage)
}
Expand Down
13 changes: 13 additions & 0 deletions core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.scheduler

import java.util.Properties
import java.util.concurrent.{CountDownLatch, TimeUnit}
import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}

import scala.annotation.meta.param
Expand Down Expand Up @@ -2849,6 +2850,18 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with TimeLi
}
}

test("SPARK-27164: RDD.countApprox on empty RDDs schedules jobs which never complete") {
val latch = new CountDownLatch(1)
val jobListener = new SparkListener {
override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {
latch.countDown()
}
}
sc.addSparkListener(jobListener)
sc.emptyRDD[Int].countApprox(10000).getFinalValue()
assert(latch.await(10, TimeUnit.SECONDS))
}

/**
* Assert that the supplied TaskSet has exactly the given hosts as its preferred locations.
* Note that this checks only the host and not the executor ID.
Expand Down