This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 755c892a7b40 [SPARK-46681][CORE] Refactor
`ExecutorFailureTracker#maxNumExecutorFailures` to avoid calculating
`defaultMaxNumExecutorFailures` when `MAX_EXECUTOR_FAILURES` is configured
755c892a7b40 is described below
commit 755c892a7b40b3ac4182c2ce12d7bea2bc411612
Author: yangjie01 <[email protected]>
AuthorDate: Fri Jan 12 02:56:09 2024 -0800
[SPARK-46681][CORE] Refactor
`ExecutorFailureTracker#maxNumExecutorFailures` to avoid calculating
`defaultMaxNumExecutorFailures` when `MAX_EXECUTOR_FAILURES` is configured
### What changes were proposed in this pull request?
This pr aims to refactor `ExecutorFailureTracker#maxNumExecutorFailures` to
avoid calculating `defaultMaxNumExecutorFailures` when `MAX_EXECUTOR_FAILURES`
is configured.
### Why are the changes needed?
Avoid unnecessary computations:
https://github.com/apache/spark/blob/bfaf9e47b35b96be01962cce0670949901e171c3/core/src/main/scala/org/apache/spark/deploy/ExecutorFailureTracker.scala#L86-L100
The result of `defaultMaxNumExecutorFailures` is calculated first, even if
MAX_EXECUTOR_FAILURES is configured now.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Pass GitHub Actions
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #44691 from LuciferYang/SPARK-46681.
Authored-by: yangjie01 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/deploy/ExecutorFailureTracker.scala | 22 ++++++++++++----------
1 file changed, 12 insertions(+), 10 deletions(-)
diff --git
a/core/src/main/scala/org/apache/spark/deploy/ExecutorFailureTracker.scala
b/core/src/main/scala/org/apache/spark/deploy/ExecutorFailureTracker.scala
index 7c7b9c60b47e..755bd83433d5 100644
--- a/core/src/main/scala/org/apache/spark/deploy/ExecutorFailureTracker.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/ExecutorFailureTracker.scala
@@ -84,18 +84,20 @@ object ExecutorFailureTracker {
// Default to twice the number of executors (twice the maximum number of
executors if dynamic
// allocation is enabled), with a minimum of 3.
def maxNumExecutorFailures(sparkConf: SparkConf): Int = {
- val effectiveNumExecutors =
- if (Utils.isStreamingDynamicAllocationEnabled(sparkConf)) {
- sparkConf.get(STREAMING_DYN_ALLOCATION_MAX_EXECUTORS)
- } else if (Utils.isDynamicAllocationEnabled(sparkConf)) {
- sparkConf.get(DYN_ALLOCATION_MAX_EXECUTORS)
- } else {
- sparkConf.get(EXECUTOR_INSTANCES).getOrElse(0)
- }
// By default, effectiveNumExecutors is Int.MaxValue if dynamic allocation
is enabled. We need
// avoid the integer overflow here.
- val defaultMaxNumExecutorFailures = math.max(3,
- if (effectiveNumExecutors > Int.MaxValue / 2) Int.MaxValue else 2 *
effectiveNumExecutors)
+ def defaultMaxNumExecutorFailures: Int = {
+ val effectiveNumExecutors =
+ if (Utils.isStreamingDynamicAllocationEnabled(sparkConf)) {
+ sparkConf.get(STREAMING_DYN_ALLOCATION_MAX_EXECUTORS)
+ } else if (Utils.isDynamicAllocationEnabled(sparkConf)) {
+ sparkConf.get(DYN_ALLOCATION_MAX_EXECUTORS)
+ } else {
+ sparkConf.get(EXECUTOR_INSTANCES).getOrElse(0)
+ }
+ math.max(3,
+ if (effectiveNumExecutors > Int.MaxValue / 2) Int.MaxValue else 2 *
effectiveNumExecutors)
+ }
sparkConf.get(MAX_EXECUTOR_FAILURES).getOrElse(defaultMaxNumExecutorFailures)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
