ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-36705][FOLLOW-UP] Fix unnecessary logWarning when PUSH_BASED_SHUFFLE_ENABLED is set to false

Browse source 
### What changes were proposed in this pull request?

Only throw logWarning when `PUSH_BASED_SHUFFLE_ENABLED` is set to true and `canDoPushBasedShuffle` is false

### Why are the changes needed?

Currently, this logWarning will still be printed out even when `PUSH_BASED_SHUFFLE_ENABLED` is set to false, which is unnecessary.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Passed existing UT.

Closes #33984 from rmcyang/SPARK-36705-follow-up.

Authored-by: Minchu Yang <minyang@minyang-mn3.linkedin.biz>
Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>
(cherry picked from commit 2d7dc7c7ce)
Signed-off-by: Mridul Muralidharan <mridulatgmail.com>
This commit is contained in:
Minchu Yang 2021-09-13 23:23:33 -05:00 committed by Mridul Muralidharan
parent 303590b3e9
commit 4a486f40cf
1 changed files with 19 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
core/src/main/scala/org/apache/spark/util/Utils.scala
View file

@ -2598,23 +2598,27 @@ private[spark] object Utils extends Logging {
* - serializer(such as KryoSerializer) supports relocation of serialized objects * - serializer(such as KryoSerializer) supports relocation of serialized objects
*/ */
def isPushBasedShuffleEnabled(conf: SparkConf): Boolean = { def isPushBasedShuffleEnabled(conf: SparkConf): Boolean = {
val serializer = Utils.classForName(conf.get(SERIALIZER)).getConstructor(classOf[SparkConf]) val pushBasedShuffleEnabled = conf.get(PUSH_BASED_SHUFFLE_ENABLED)
.newInstance(conf).asInstanceOf[Serializer] if (pushBasedShuffleEnabled) {
val canDoPushBasedShuffle = val serializer = Utils.classForName(conf.get(SERIALIZER)).getConstructor(classOf[SparkConf])
conf.get(PUSH_BASED_SHUFFLE_ENABLED) && .newInstance(conf).asInstanceOf[Serializer]
(conf.get(IS_TESTING).getOrElse(false) || val canDoPushBasedShuffle = conf.get(IS_TESTING).getOrElse(false) ||
(conf.get(SHUFFLE_SERVICE_ENABLED) && (conf.get(SHUFFLE_SERVICE_ENABLED) &&
conf.get(SparkLauncher.SPARK_MASTER, null) == "yarn" && conf.get(SparkLauncher.SPARK_MASTER, null) == "yarn" &&
// TODO: [SPARK-36744] needs to support IO encryption for push-based shuffle // TODO: [SPARK-36744] needs to support IO encryption for push-based shuffle
!conf.get(IO_ENCRYPTION_ENABLED) && !conf.get(IO_ENCRYPTION_ENABLED) &&
serializer.supportsRelocationOfSerializedObjects)) serializer.supportsRelocationOfSerializedObjects)
if (!canDoPushBasedShuffle) { if (!canDoPushBasedShuffle) {
logWarning("Push-based shuffle can only be enabled when the application is submitted" + logWarning("Push-based shuffle can only be enabled when the application is submitted " +
"to run in YARN mode, with external shuffle service enabled, IO encryption disabled, and" + "to run in YARN mode, with external shuffle service enabled, IO encryption disabled, " +
"relocation of serialized objects supported.") "and relocation of serialized objects supported.")
}
canDoPushBasedShuffle
} else {
false
} }
canDoPushBasedShuffle
} }
/** /**