Skip to content

[SPARK-33933][SQL] Materialize BroadcastQueryStage first to avoid broadcast timeout in AQE #30998

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
zhongyu09 wants to merge 4 commits into from
Closed

[SPARK-33933][SQL] Materialize BroadcastQueryStage first to avoid broadcast timeout in AQE #30998

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
25be218
SPARK-33933: materialize BroadcastQueryState first to avoid broadcast…
Jan 3, 2021
c49065c
update comment
Jan 4, 2021
83e3e4e
reduce UT time within 5 sec
Jan 5, 2021
7cbeb14
move UT from BroadcastJoinSuite to AdaptiveQueryExecSuite
Jan 6, 2021
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
11 changes: 10 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,8 +189,17 @@ case class AdaptiveSparkPlanExec(
stagesToReplace = result.newStages ++ stagesToReplace
executionId.foreach(onUpdatePlan(_, result.newStages.map(_.plan)))

// SPARK-33933: we should submit tasks of broadcast stages first, to avoid waiting
// for tasks to be scheduled and leading to broadcast timeout.
val reorderedNewStages = result.newStages
.sortWith {
case (_: BroadcastQueryStageExec, _: BroadcastQueryStageExec) => false
case (_: BroadcastQueryStageExec, _) => true
case _ => false
}

// Start materialization of all new stages and fail fast if any stages failed eagerly
result.newStages.foreach { stage =>
reorderedNewStages.foreach { stage =>
try {
stage.materialize().onComplete { res =>
if (res.isSuccess) {
Expand Down
24 changes: 24 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1431,4 +1431,28 @@ class AdaptiveQueryExecSuite
}
}
}

test("SPARK-33933: AQE broadcast should not timeout with slow map tasks") {
Copy link
Contributor

@LuciferYang LuciferYang Jan 8, 2021

Choose a reason for hiding this comment

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

It seems that this case fails a little frequently：

@zhongyu09 @cloud-fan @viirya

Copy link
Member

@viirya viirya Jan 8, 2021

Choose a reason for hiding this comment

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

Hmm, so it is still flaky.

Copy link
Contributor Author

@zhongyu09 zhongyu09 Jan 8, 2021

Choose a reason for hiding this comment

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

I try the test tens of times and the test failed twice. As we discussed, the guarantee is not strict since the submit of broadcast job and shuffle map job are in different thread. So there's still risk for the shuffle map job schedule earlier before broadcast job. I wonder should we need to remove the UT until we thorough resolve the issue.

CC @cloud-fan @dongjoon-hyun

Copy link
Contributor

@LuciferYang LuciferYang Jan 8, 2021

Choose a reason for hiding this comment

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

I run SPARK-33933: AQE broadcast should not timeout with slow map tasks in local 5 times and all failed as follow:

- SPARK-33933: AQE broadcast should not timeout with slow map tasks *** FAILED ***
  1751 was not greater than 2000 (AdaptiveQueryExecSuite.scala:1454)

Copy link
Contributor

@LuciferYang LuciferYang Jan 8, 2021

Choose a reason for hiding this comment

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

Hmm, so it is still flaky.

yes

Copy link
Contributor

@cloud-fan cloud-fan Jan 8, 2021

Choose a reason for hiding this comment

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

the spark conf changed to local[2] and so the running times are faster than before.

This shows the test is unreliable...

Checking the Spark jobs submission order should be easy to do and fast to run, and with retry it should be unlikely to fail. It's better to check stage submission order directly, if we can figure out how to do it.

Copy link
Contributor Author

@zhongyu09 zhongyu09 Jan 8, 2021

Choose a reason for hiding this comment

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

Yes I know, the failure reported by @LuciferYang is easy to solve.
But the question is, the jobs submission order may be not correct, like the test in #31084.

Copy link
Contributor Author

@zhongyu09 zhongyu09 Jan 8, 2021

Choose a reason for hiding this comment

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

@cloud-fan You mean with retry it should be unlikely to fail to solve the edge case?

Copy link
Contributor Author

@zhongyu09 zhongyu09 Jan 8, 2021

Choose a reason for hiding this comment

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

I am trying to create a new PR. Sorry for the inconvenience.

Copy link
Contributor Author

@zhongyu09 zhongyu09 Jan 9, 2021

Choose a reason for hiding this comment

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

We can get the stage submission time using SparkListeneer. But after trying serval times, the stage submission time is not stable thus the UT cannot always passed. I suggest to remove the UT before we completely solve the issue. #31099

val broadcastTimeoutInSec = 1
val df = spark.sparkContext.parallelize(Range(0, 100), 100)
.flatMap(x => {
Thread.sleep(20)
for (i <- Range(0, 100)) yield (x % 26, x % 10)
}).toDF("index", "pv")
val dim = Range(0, 26).map(x => (x, ('a' + x).toChar.toString))
.toDF("index", "name")
val testDf = df.groupBy("index")
.agg(sum($"pv").alias("pv"))
.join(dim, Seq("index"))
withSQLConf(SQLConf.BROADCAST_TIMEOUT.key -> broadcastTimeoutInSec.toString,
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
val startTime = System.currentTimeMillis()
val result = testDf.collect()
val queryTime = System.currentTimeMillis() - startTime
assert(result.length == 26)
// make sure the execution time is large enough
assert(queryTime > (broadcastTimeoutInSec + 1) * 1000)
}
}

}