[incubator-celeborn] branch branch-0.3 updated: [CELEBORN-801] Warn when local shuffle reader is enabled

[zh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zhouky pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/branch-0.3 by this push:
     new 251730b62 [CELEBORN-801] Warn when local shuffle reader is enabled
251730b62 is described below

commit 251730b62ab29d0fb23da43cc7ef97b35b08c94f
Author: Cheng Pan <ch...@apache.org>
AuthorDate: Mon Jul 17 16:43:50 2023 +0800

    [CELEBORN-801] Warn when local shuffle reader is enabled
    
    ### What changes were proposed in this pull request?
    
    Warn when local shuffle reader is enabled.
    
    ```
    Detected spark.sql.adaptive.localShuffleReader.enabled (default is true) is enabled,
    it's highly recommended to disable it when use Celeborn as Remote Shuffle Service to
    avoid performance degradation.
    ```
    
    ### Why are the changes needed?
    
    When local shuffle reader is enabled, the reduce task may read shuffle data in by map id, which is not match the Celeborn shuffle data clustering model, then cause extremely bad shuffle read performance.
    
    ### Does this PR introduce _any_ user-facing change?
    
    Yes, user would see warning message from Driver log when `spark.sql.adaptive.localShuffleReader.enabled` is true.
    
    ### How was this patch tested?
    
    Review.
    
    Closes #1721 from pan3793/CELEBORN-801.
    
    Authored-by: Cheng Pan <ch...@apache.org>
    Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
    (cherry picked from commit 1ec4f4a9f558f2686351c4765be9b7c5621ef778)
    Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
 .../apache/spark/shuffle/celeborn/SparkShuffleManager.java  | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
index f9ff30d04..9695fd8ac 100644
--- a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
+++ b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
@@ -41,9 +41,12 @@ public class SparkShuffleManager implements ShuffleManager {
 
   private static final Logger logger = LoggerFactory.getLogger(SparkShuffleManager.class);
 
-  private static final String sortShuffleManagerName =
+  private static final String SORT_SHUFFLE_MANAGER_NAME =
       "org.apache.spark.shuffle.sort.SortShuffleManager";
 
+  private static final String LOCAL_SHUFFLE_READER_KEY =
+      "spark.sql.adaptive.localShuffleReader.enabled";
+
   private final SparkConf conf;
   private final CelebornConf celebornConf;
   private final int cores;
@@ -61,6 +64,12 @@ public class SparkShuffleManager implements ShuffleManager {
   private AtomicInteger pusherIdx = new AtomicInteger(0);
 
   public SparkShuffleManager(SparkConf conf) {
+    if (conf.getBoolean(LOCAL_SHUFFLE_READER_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.",
+          LOCAL_SHUFFLE_READER_KEY);
+    }
     this.conf = conf;
     this.celebornConf = SparkUtils.fromSparkConf(conf);
     this.cores = executorCores(conf);
@@ -85,7 +94,7 @@ public class SparkShuffleManager implements ShuffleManager {
       synchronized (this) {
         if (_sortShuffleManager == null) {
           _sortShuffleManager =
-              SparkUtils.instantiateClass(sortShuffleManagerName, conf, isDriver());
+              SparkUtils.instantiateClass(SORT_SHUFFLE_MANAGER_NAME, conf, isDriver());
         }
       }
     }