Skip to content

[KYUUBI #5136][Bug] Spark App may hang forever if FinalStageResourceManager killed all executors #5141

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
zhouyifan279 wants to merge 5 commits into from
Closed

[KYUUBI #5136][Bug] Spark App may hang forever if FinalStageResourceManager killed all executors #5141

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9dcbc78
[KYUUBI #5136][Bug] Spark App may hang forever if FinalStageResourceM…
zhouyifan279 Aug 9, 2023
12687ee
[KYUUBI #5136][Bug] Spark App may hang forever if FinalStageResourceM…
zhouyifan279 Aug 14, 2023
5f3ca1d
[KYUUBI #5136][Bug] Spark App may hang forever if FinalStageResourceM…
zhouyifan279 Aug 15, 2023
ea8f247
Add comment
zhouyifan279 Aug 15, 2023
c4403ee
assert adjustedTargetExecutors == 1
zhouyifan279 Aug 16, 2023
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
Prev Previous commit
assert adjustedTargetExecutors == 1
  • Loading branch information
@zhouyifan279
zhouyifan279 committed Aug 16, 2023
commit c4403eefa6c80bab4a160b747a756329b1193307
2 changes: 1 addition & 1 deletion ...bi-extension-spark-3-3/src/main/scala/org/apache/kyuubi/sql/KyuubiSparkSQLExtension.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ class KyuubiSparkSQLExtension extends (SparkSessionExtensions => Unit) {
extensions.injectOptimizerRule(ForcedMaxOutputRowsRule)
extensions.injectPlannerStrategy(MaxScanStrategy)

extensions.injectQueryStagePrepRule(FinalStageResourceManager)
extensions.injectQueryStagePrepRule(FinalStageResourceManager(_))
extensions.injectQueryStagePrepRule(InjectCustomResourceProfile)
}
}
22 changes: 12 additions & 10 deletions ...i-extension-spark-3-3/src/main/scala/org/apache/spark/sql/FinalStageResourceManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

import org.apache.spark.{ExecutorAllocationClient, MapOutputTrackerMaster, SparkContext, SparkEnv}
import org.apache.spark.internal.Logging
import org.apache.spark.resource.ResourceProfile
import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
import org.apache.spark.sql.catalyst.rules.Rule
Expand Down Expand Up @@ -217,7 +218,7 @@ case class FinalStageResourceManager(session: SparkSession)
countFailures = false,
force = false)
Copy link
Contributor

@ulysses-you ulysses-you Aug 8, 2023

Choose a reason for hiding this comment

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

Shall we call client.requestTotalExecutors to adjust the target executor if targetExecutors < draTargetExecutors after kill executors ?

We might face a issue similar with apache/spark#19048

Copy link
Contributor

@ulysses-you ulysses-you Aug 8, 2023

Choose a reason for hiding this comment

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

IMO, we just need to tune adjustTargetNumExecutors from true to false and call client.requestTotalExecutors to ensure the final target executor is expected.

Copy link
Contributor Author

@zhouyifan279 zhouyifan279 Aug 8, 2023

Choose a reason for hiding this comment

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

Agree. But build parameters of client.requestTotalExecutors involves many details.
So I prefer to wait ExecutorAllocationManager to call client.requestTotalExecutors.

As long as we did not call killExecutors with adjustTargetNumExecutors = true, ExecutorAllocationManager should be able to manage target executor num correctly.


getAdjustedTargetExecutors(sc, executorAllocationClient)
FinalStageResourceManager.getAdjustedTargetExecutors(sc)
.filter(_ < targetExecutors).foreach { adjustedExecutors =>
val delta = targetExecutors - adjustedExecutors
logInfo(s"Target executors after kill ($adjustedExecutors) is lower than required " +
Expand All @@ -226,10 +227,16 @@ case class FinalStageResourceManager(session: SparkSession)
}
}

private def getAdjustedTargetExecutors(
sc: SparkContext,
executorAllocationClient: ExecutorAllocationClient): Option[Int] = {
executorAllocationClient match {
@transient private val queryStageOptimizerRules: Seq[Rule[SparkPlan]] = Seq(
OptimizeSkewInRebalancePartitions,
CoalesceShufflePartitions(session),
OptimizeShuffleWithLocalRead)
}

object FinalStageResourceManager extends Logging {

private[sql] def getAdjustedTargetExecutors(sc: SparkContext): Option[Int] = {
sc.schedulerBackend match {
case schedulerBackend: CoarseGrainedSchedulerBackend =>
try {
val field = classOf[CoarseGrainedSchedulerBackend]
Expand All @@ -249,11 +256,6 @@ case class FinalStageResourceManager(session: SparkSession)
case _ => None
}
}

@transient private val queryStageOptimizerRules: Seq[Rule[SparkPlan]] = Seq(
OptimizeSkewInRebalancePartitions,
CoalesceShufflePartitions(session),
OptimizeShuffleWithLocalRead)
}

trait FinalRebalanceStageHelper extends AdaptiveSparkPlanHelper {
Expand Down
1 change: 1 addition & 0 deletions ...ension-spark-3-3/src/test/scala/org/apache/spark/sql/FinalStageResourceManagerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ class FinalStageResourceManagerSuite extends KyuubiSparkSQLExtensionTest {
sql(
"CREATE TABLE final_stage AS SELECT id, count(*) as num FROM (SELECT 0 id) GROUP BY id")
}
Copy link
Contributor

@ulysses-you ulysses-you Aug 16, 2023

Choose a reason for hiding this comment

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

shall we getAdjustedTargetExecutors at the end to make sure the target executor number is 1 ?

Copy link
Contributor Author

@zhouyifan279 zhouyifan279 Aug 16, 2023

Choose a reason for hiding this comment

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

done

assert(FinalStageResourceManager.getAdjustedTargetExecutors(spark.sparkContext).get == 1)
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion ...bi-extension-spark-3-4/src/main/scala/org/apache/kyuubi/sql/KyuubiSparkSQLExtension.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ class KyuubiSparkSQLExtension extends (SparkSessionExtensions => Unit) {
extensions.injectOptimizerRule(ForcedMaxOutputRowsRule)
extensions.injectPlannerStrategy(MaxScanStrategy)

extensions.injectQueryStagePrepRule(FinalStageResourceManager)
extensions.injectQueryStagePrepRule(FinalStageResourceManager(_))
extensions.injectQueryStagePrepRule(InjectCustomResourceProfile)
}
}
22 changes: 12 additions & 10 deletions ...i-extension-spark-3-4/src/main/scala/org/apache/spark/sql/FinalStageResourceManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

import org.apache.spark.{ExecutorAllocationClient, MapOutputTrackerMaster, SparkContext, SparkEnv}
import org.apache.spark.internal.Logging
import org.apache.spark.resource.ResourceProfile
import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
import org.apache.spark.sql.catalyst.rules.Rule
Expand Down Expand Up @@ -220,7 +221,7 @@ case class FinalStageResourceManager(session: SparkSession)
countFailures = false,
force = false)

getAdjustedTargetExecutors(sc, executorAllocationClient)
FinalStageResourceManager.getAdjustedTargetExecutors(sc)
.filter(_ < targetExecutors).foreach { adjustedExecutors =>
val delta = targetExecutors - adjustedExecutors
logInfo(s"Target executors after kill ($adjustedExecutors) is lower than required " +
Expand All @@ -229,10 +230,16 @@ case class FinalStageResourceManager(session: SparkSession)
}
}

private def getAdjustedTargetExecutors(
sc: SparkContext,
executorAllocationClient: ExecutorAllocationClient): Option[Int] = {
executorAllocationClient match {
@transient private val queryStageOptimizerRules: Seq[Rule[SparkPlan]] = Seq(
OptimizeSkewInRebalancePartitions,
CoalesceShufflePartitions(session),
OptimizeShuffleWithLocalRead)
}

object FinalStageResourceManager extends Logging {

private[sql] def getAdjustedTargetExecutors(sc: SparkContext): Option[Int] = {
sc.schedulerBackend match {
case schedulerBackend: CoarseGrainedSchedulerBackend =>
try {
val field = classOf[CoarseGrainedSchedulerBackend]
Expand All @@ -252,11 +259,6 @@ case class FinalStageResourceManager(session: SparkSession)
case _ => None
}
}

@transient private val queryStageOptimizerRules: Seq[Rule[SparkPlan]] = Seq(
OptimizeSkewInRebalancePartitions,
CoalesceShufflePartitions(session),
OptimizeShuffleWithLocalRead)
}

trait FinalRebalanceStageHelper extends AdaptiveSparkPlanHelper {
Expand Down
1 change: 1 addition & 0 deletions ...ension-spark-3-4/src/test/scala/org/apache/spark/sql/FinalStageResourceManagerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ class FinalStageResourceManagerSuite extends KyuubiSparkSQLExtensionTest {
sql(
"CREATE TABLE final_stage AS SELECT id, count(*) as num FROM (SELECT 0 id) GROUP BY id")
}
assert(FinalStageResourceManager.getAdjustedTargetExecutors(spark.sparkContext).get == 1)
}
}
}
Expand Down