This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new dfde5de872 [CELEBORN-2451] Make DRA safe under AUTO fallback by tying
supportsReliableStorage to fallback policy
dfde5de872 is described below
commit dfde5de8725f844bf07d8fcd3ef98103b980d974
Author: Venkata Krishnan Sowrirajan <[email protected]>
AuthorDate: Thu Oct 22 20:51:08 2026 +0800
[CELEBORN-2451] Make DRA safe under AUTO fallback by tying
supportsReliableStorage to fallback policy
### What changes were proposed in this pull request?
- Tie `supportsReliableStorage()` to the fallback policy: `true` only for
`NEVER`, `false` for `AUTO`/`ALWAYS` (was keyed on deprecated
`forceFallback`, so it returned `true` under `AUTO`).
- Make `SparkShuffleManager` DRA warnings policy-aware; log the
`registerShuffle` fallback at `error` only when shuffle tracking is also
off.
### Why are the changes needed?
- Fallback is per-shuffle, but `supportsReliableStorage()` is one global
flag.
Under `AUTO` a shuffle can fall back to local-disk `SortShuffleManager`;
that output dies with a DRA-reclaimed executor, causing
`FetchFailedException`.
Nothing prevents the unsafe config today.
- Reporting `false` under `AUTO`/`ALWAYS` lets Spark's
`ExecutorAllocationManager`
fail fast unless a shuffle service, shuffle tracking, or decommissioning
is on,
so `AUTO` is safe with `shuffleTracking.enabled=true`; `NEVER` keeps the
fast path.
- Trade-off: under `AUTO`, `reliableStorage=false` makes Spark recompute a
lost
executor's map stages even though Celeborn still holds the data; `NEVER`
avoids
this. Removing it needs a Spark-side change (per-shuffle reliable-storage
granularity), tracked as a follow-up.
### Does this PR resolve a correctness bug?
- [x] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes: with `AUTO` (default), DRA without shuffle service and without
shuffle
tracking now fails fast at startup. Enable tracking or set `NEVER`.
### How was this patch tested?
- New `testReliableStorageFollowsFallbackPolicy`;
`-Dtest=SparkShuffleManagerSuite` passes (5/5), spotless clean.
Closes #3834 from venkata91/dra-auto-fallback-reliable-storage.
Authored-by: Venkata Krishnan Sowrirajan <[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
---
README.md | 17 +++++-
.../shuffle/celeborn/CelebornShuffleDataIO.java | 6 +-
.../shuffle/celeborn/SparkShuffleManager.java | 60 ++++++++++++++-----
.../celeborn/CelebornShuffleManagerSuite.scala | 69 +++++++++++++++++++++-
docs/deploy.md | 17 +++++-
5 files changed, 148 insertions(+), 21 deletions(-)
diff --git a/README.md b/README.md
index 2f14f5b668..f2f2812711 100644
--- a/README.md
+++ b/README.md
@@ -320,14 +320,27 @@ spark.sql.adaptive.skewJoin.enabled true
# Support Spark Dynamic Resource Allocation
# Required Spark version >= 3.5.0
spark.shuffle.sort.io.plugin.class
org.apache.spark.shuffle.celeborn.CelebornShuffleDataIO
-# Required Spark version >= 3.4.0, highly recommended to disable
-spark.dynamicAllocation.shuffleTracking.enabled false
+# Available since Spark 3.0; enabled by default since Spark 3.4.
+# With AUTO (default) or ALWAYS fallback, keep tracking enabled to protect
local-disk shuffle data.
+spark.dynamicAllocation.shuffleTracking.enabled true
# Support ShuffleManager when defined in user jars
# Required Spark version < 4.0.0 or without SPARK-45762, highly recommended to
false for ShuffleManager in user-defined jar specified by --jars or spark.jars
spark.executor.userClassPathFirst false
```
+With DRA enabled, `AUTO` and `ALWAYS` fallback policies require shuffle
tracking unless an external
+shuffle service or Spark shuffle decommissioning is configured. Shuffle
decommissioning requires
+`spark.decommission.enabled=true` and
`spark.storage.decommission.shuffleBlocks.enabled=true`;
+also enable `spark.storage.decommission.enabled` to migrate stored blocks.
+To keep all shuffle data on Celeborn, set
`spark.celeborn.client.spark.shuffle.fallback.policy=NEVER`.
+With `NEVER`, the application fails rather than falling back when Celeborn
cannot serve a shuffle.
+Shuffle tracking can then be disabled to release idle executors promptly, but
only when Spark's
+dynamic allocation check still passes without it: either use Spark 3.5 or
later with
+`spark.shuffle.sort.io.plugin.class=org.apache.spark.shuffle.celeborn.CelebornShuffleDataIO`,
or
+enable Spark shuffle decommissioning as described above. Otherwise, for
example on Spark 3.4 without
+an external shuffle service or shuffle decommissioning, Spark rejects dynamic
allocation at startup.
+
Shuffle OSS storage related configurations:
```properties
# If you are using Celeborn for shuffle OSS storage, these settings will be
needed.
diff --git
a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/CelebornShuffleDataIO.java
b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/CelebornShuffleDataIO.java
index 45b00320f5..acc3d3598e 100644
---
a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/CelebornShuffleDataIO.java
+++
b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/CelebornShuffleDataIO.java
@@ -27,6 +27,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.protocol.FallbackPolicy;
public class CelebornShuffleDataIO implements ShuffleDataIO {
@@ -58,7 +59,10 @@ class CelebornShuffleDriverComponents extends
LocalDiskShuffleDriverComponents {
private final boolean supportsReliableStorage;
public CelebornShuffleDriverComponents(CelebornConf celebornConf) {
- this.supportsReliableStorage = !celebornConf.shuffleForceFallbackEnabled();
+ // Reliable only under NEVER. AUTO/ALWAYS may fall back to local-disk
shuffle, whose output
+ // dies with the executor, so report false to stop DRA from reclaiming
executors that hold it.
+ this.supportsReliableStorage =
+ FallbackPolicy.NEVER.equals(celebornConf.sparkShuffleFallbackPolicy());
}
// Omitting @Override annotation to avoid compile error before Spark 3.5.0
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 71aae81f31..8d4749945b 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
@@ -23,6 +23,7 @@ import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.spark.*;
+import org.apache.spark.internal.config.ConfigEntry;
import org.apache.spark.internal.config.package$;
import org.apache.spark.launcher.SparkLauncher;
import org.apache.spark.rdd.DeterministicLevel;
@@ -36,6 +37,7 @@ import org.apache.celeborn.client.LifecycleManager;
import org.apache.celeborn.client.ShuffleClient;
import org.apache.celeborn.client.security.CryptoHandler;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.protocol.FallbackPolicy;
import org.apache.celeborn.common.protocol.ShuffleMode;
import org.apache.celeborn.reflect.DynMethods;
import org.apache.celeborn.spark.FailedShuffleCleaner;
@@ -58,6 +60,22 @@ public class SparkShuffleManager implements ShuffleManager {
private static final String SORT_SHUFFLE_MANAGER_NAME =
"org.apache.spark.shuffle.sort.SortShuffleManager";
+ // Decommissioning configs do not exist in Spark 3.0. Read Spark's entries
to preserve its
+ // defaults.
+ private static final ConfigEntry<Object> DECOMMISSION_ENABLED =
+ DynMethods.builder("DECOMMISSION_ENABLED")
+ .impl(package$.class)
+ .orNoop()
+ .build(package$.MODULE$)
+ .invoke();
+
+ private static final ConfigEntry<Object>
STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED =
+ DynMethods.builder("STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED")
+ .impl(package$.class)
+ .orNoop()
+ .build(package$.MODULE$)
+ .invoke();
+
private static final boolean COLUMNAR_SHUFFLE_CLASSES_PRESENT;
static {
@@ -110,6 +128,20 @@ public class SparkShuffleManager implements ShuffleManager
{
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(
@@ -117,21 +149,18 @@ public class SparkShuffleManager implements
ShuffleManager {
+ "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 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());
}
SparkCommonUtils.validateAttemptConfig(conf);
this.conf = conf;
this.isDriver = isDriver;
- this.celebornConf = SparkUtils.fromSparkConf(conf);
this.cores = executorCores(conf);
this.fallbackPolicyRunner = new
CelebornShuffleFallbackPolicyRunner(celebornConf);
this.sendBufferPoolCheckInterval =
celebornConf.clientPushSendBufferPoolExpireCheckInterval();
@@ -220,13 +249,14 @@ public class SparkShuffleManager implements
ShuffleManager {
lifecycleManager.shuffleCount().increment();
if (fallbackPolicyRunner.applyFallbackPolicies(dependency,
lifecycleManager)) {
- if (conf.getBoolean("spark.dynamicAllocation.enabled", false)
- && !conf.getBoolean("spark.shuffle.service.enabled", false)) {
+ if (isUnsafeDraFallback(conf)) {
logger.error(
"DRA is enabled but we fallback to vanilla Spark
SortShuffleManager for "
- + "shuffle: {} due to fallback policy. It may cause block can
not found when reducer "
- + "task fetch data.",
- shuffleId);
+ + "shuffle: {} due to fallback policy, and {} is disabled
while shuffle "
+ + "decommissioning is not enabled. Fallback shuffle blocks may
be lost when idle "
+ + "executors are reclaimed.",
+ shuffleId,
+ package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED().key());
} else {
logger.warn("Fallback to vanilla Spark SortShuffleManager for shuffle:
{}", shuffleId);
}
diff --git
a/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleManagerSuite.scala
b/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleManagerSuite.scala
index bded509413..2a3daa0f85 100644
---
a/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleManagerSuite.scala
+++
b/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleManagerSuite.scala
@@ -17,7 +17,7 @@
package org.apache.spark.shuffle.celeborn
-import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.{SparkConf, SparkContext, SparkVersionUtil}
import org.apache.spark.sql.internal.SQLConf
import org.junit
import org.junit.Assert
@@ -92,6 +92,73 @@ class SparkShuffleManagerSuite extends Logging {
sc.stop()
}
+ @junit.Test
+ def testReliableStorageFollowsFallbackPolicy(): Unit = {
+ // supportsReliableStorage must be true only for NEVER; AUTO/ALWAYS may
fall back to local-disk
+ // shuffle, so reporting reliable storage would let DRA reclaim executors
holding fallback data.
+ def reliable(policy: String): Boolean = {
+ val conf = new SparkConf(false)
+ .set(s"spark.${CelebornConf.SPARK_SHUFFLE_FALLBACK_POLICY.key}",
policy)
+ new CelebornShuffleDataIO(conf).driver()
+
.asInstanceOf[CelebornShuffleDriverComponents].supportsReliableStorage()
+ }
+ Assert.assertTrue(reliable("NEVER"))
+ Assert.assertFalse(reliable("AUTO"))
+ Assert.assertFalse(reliable("ALWAYS"))
+ Assert.assertFalse(
+ new CelebornShuffleDataIO(new SparkConf(false)).driver()
+
.asInstanceOf[CelebornShuffleDriverComponents].supportsReliableStorage())
+ }
+
+ @junit.Test
+ def testUnsafeDraFallback(): Unit = {
+ for {
+ dra <- Seq(false, true)
+ shuffleService <- Seq(false, true)
+ tracking <- Seq(false, true)
+ decommission <- Seq(false, true)
+ shuffleBlocks <- Seq(false, true)
+ } {
+ val conf = new SparkConf(false)
+ .set("spark.dynamicAllocation.enabled", dra.toString)
+ .set("spark.shuffle.service.enabled", shuffleService.toString)
+ .set("spark.dynamicAllocation.shuffleTracking.enabled",
tracking.toString)
+ .set("spark.decommission.enabled", decommission.toString)
+ .set("spark.storage.decommission.shuffleBlocks.enabled",
shuffleBlocks.toString)
+ val decommissionSupported = SparkVersionUtil.isGreaterThan(3, 0)
+ val expected = dra && !shuffleService && !tracking &&
+ !(decommissionSupported && decommission && shuffleBlocks)
+ Assert.assertEquals(
+ conf.toDebugString,
+ expected,
+ SparkShuffleManager.isUnsafeDraFallback(conf))
+ }
+ }
+
+ @junit.Test
+ def testUnsafeDraFallbackUsesSparkDefaults(): Unit = {
+ val conf = new SparkConf(false)
+ .set("spark.dynamicAllocation.enabled", "true")
+ .set("spark.shuffle.service.enabled", "false")
+ .set("spark.dynamicAllocation.shuffleTracking.enabled", "false")
+ Assert.assertTrue(SparkShuffleManager.isUnsafeDraFallback(conf))
+
+ conf.set("spark.decommission.enabled", "true")
+ // Spark 3.4 changed shuffle-block decommissioning's default from false to
true.
+ Assert.assertEquals(
+ !SparkVersionUtil.isGreaterThan(3, 3),
+ SparkShuffleManager.isUnsafeDraFallback(conf))
+
+ conf.remove("spark.decommission.enabled")
+ conf.set("spark.storage.decommission.shuffleBlocks.enabled", "true")
+ Assert.assertTrue(SparkShuffleManager.isUnsafeDraFallback(conf))
+
+ conf.remove("spark.dynamicAllocation.shuffleTracking.enabled")
+ Assert.assertEquals(
+ !SparkVersionUtil.isGreaterThan(3, 3),
+ SparkShuffleManager.isUnsafeDraFallback(conf))
+ }
+
@junit.Test
def testWrongSparkConfMaxAttemptLimit(): Unit = {
val conf = new SparkConf().setIfMissing("spark.master", "local")
diff --git a/docs/deploy.md b/docs/deploy.md
index fef96b6146..9e5835a9a2 100644
--- a/docs/deploy.md
+++ b/docs/deploy.md
@@ -208,14 +208,27 @@ spark.sql.adaptive.skewJoin.enabled true
# Support Spark Dynamic Resource Allocation
# Required Spark version >= 3.5.0
spark.shuffle.sort.io.plugin.class
org.apache.spark.shuffle.celeborn.CelebornShuffleDataIO
-# Required Spark version >= 3.4.0, highly recommended to disable
-spark.dynamicAllocation.shuffleTracking.enabled false
+# Available since Spark 3.0; enabled by default since Spark 3.4.
+# With AUTO (default) or ALWAYS fallback, keep tracking enabled to protect
local-disk shuffle data.
+spark.dynamicAllocation.shuffleTracking.enabled true
# Support ShuffleManager when defined in user jars
# Required Spark version < 4.0.0 or without SPARK-45762, highly recommended to
false for ShuffleManager in user-defined jar specified by --jars or spark.jars
spark.executor.userClassPathFirst false
```
+With DRA enabled, `AUTO` and `ALWAYS` fallback policies require shuffle
tracking unless an external
+shuffle service or Spark shuffle decommissioning is configured. Shuffle
decommissioning requires
+`spark.decommission.enabled=true` and
`spark.storage.decommission.shuffleBlocks.enabled=true`;
+also enable `spark.storage.decommission.enabled` to migrate stored blocks.
+To keep all shuffle data on Celeborn, set
`spark.celeborn.client.spark.shuffle.fallback.policy=NEVER`.
+With `NEVER`, the application fails rather than falling back when Celeborn
cannot serve a shuffle.
+Shuffle tracking can then be disabled to release idle executors promptly, but
only when Spark's
+dynamic allocation check still passes without it: either use Spark 3.5 or
later with
+`spark.shuffle.sort.io.plugin.class=org.apache.spark.shuffle.celeborn.CelebornShuffleDataIO`,
or
+enable Spark shuffle decommissioning as described above. Otherwise, for
example on Spark 3.4 without
+an external shuffle service or shuffle decommissioning, Spark rejects dynamic
allocation at startup.
+
## Deploy Flink client
**Important: Only Flink batch jobs are supported for now. Due to the Shuffle
Service in Flink is cluster-granularity, if you want to use Celeborn in a
session cluster, it will not be able to submit both streaming and batch job to
the same cluster. We plan to get rid of this restriction for Hybrid Shuffle
mode in a future release.**
