SteNicholas commented on code in PR #3834:
URL: https://github.com/apache/celeborn/pull/3834#discussion_r4258712688
##########
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:
@venkata91, qualify the recommendation to disable shuffle tracking. `NEVER`
alone does not make shuffle tracking unnecessary for Spark's dynamic-allocation
validation. For example, with DRA enabled on Spark 3.4 and neither an external
shuffle service nor shuffle decommissioning configured, following this warning
makes the application fail at startup. On Spark 3.5+, disabling tracking also
requires an IO plugin that reports reliable storage, such as
`CelebornShuffleDataIO`, unless one of those other mechanisms is configured.
Please guard this recommendation with the required capability or state these
prerequisites explicitly, consistent with the README and deployment guide.
--
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]
