venkata91 commented on code in PR #3834:
URL: https://github.com/apache/celeborn/pull/3834#discussion_r4321991100
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java:
##########
@@ -110,28 +127,47 @@ private Optional<CryptoHandler> getCryptoHandler() {
return cryptoHandler;
}
+ private static boolean isDecommissionShuffleBlocksEnabled(SparkConf conf) {
+ return DECOMMISSION_ENABLED != null
+ && STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED != null
+ && (Boolean) conf.get(DECOMMISSION_ENABLED)
+ && (Boolean) conf.get(STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED);
+ }
+
+ static boolean isUnsafeDraFallback(SparkConf conf) {
+ return conf.getBoolean("spark.dynamicAllocation.enabled", false)
+ && !conf.getBoolean("spark.shuffle.service.enabled", false)
+ && !(Boolean)
conf.get(package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED())
+ && !isDecommissionShuffleBlocksEnabled(conf);
+ }
+
public SparkShuffleManager(SparkConf conf, boolean isDriver) {
if (conf.getBoolean(SQLConf.LOCAL_SHUFFLE_READER_ENABLED().key(), true)) {
logger.warn(
"Detected {} (default is true) is enabled, it's highly recommended
to disable it when "
+ "use Celeborn as Remote Shuffle Service to avoid performance
degradation.",
SQLConf.LOCAL_SHUFFLE_READER_ENABLED().key());
}
- if ((Boolean)
conf.get(package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED())) {
- String key =
package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED().key();
- Boolean defaultValue =
- (Boolean)
package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED().defaultValue().get();
+ this.celebornConf = SparkUtils.fromSparkConf(conf);
+ boolean shuffleTrackingEnabled =
+ (Boolean)
conf.get(package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED());
+ boolean neverFallback =
FallbackPolicy.NEVER.equals(celebornConf.sparkShuffleFallbackPolicy());
+ if (!neverFallback && isUnsafeDraFallback(conf)) {
logger.warn(
- "Detected {} (default is {}) is enabled, "
- + "it's highly recommended to disable it when use Celeborn as
Remote Shuffle Service "
- + "to avoid performance degradation.",
- key,
- defaultValue);
+ "DRA is enabled without the external shuffle service and fallback
policy is not NEVER, "
+ + "but {} is disabled and shuffle decommissioning is not
enabled. Enable tracking so fallback "
+ + "shuffle output is not lost when idle executors are
reclaimed.",
+ package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED().key());
+ } else if (neverFallback && shuffleTrackingEnabled) {
+ // Under NEVER all shuffle stays on Celeborn, so tracking only delays
releasing executors.
+ logger.warn(
+ "{} is enabled with fallback policy NEVER; it is unnecessary and
delays releasing idle "
+ + "executors, so it is recommended to disable it.",
Review Comment:
Thanks @SteNicholas. Rather than guarding the hint in code, I removed the
runtime recommendation to disable shuffle tracking under `NEVER` entirely
(000ec58). It was only a performance hint, and its prerequisites are subtle:
without an external shuffle service, Spark 3.5+ needs `CelebornShuffleDataIO`
configured, or shuffle decommissioning must be enabled, and otherwise Spark
rejects DRA at startup. The README and deployment guide now state these
prerequisites explicitly. The startup log now only warns when `AUTO`/`ALWAYS`
fallback is unsafe under DRA (no shuffle service, no tracking, and no shuffle
decommissioning).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
