apache/celeborn
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[CELEBORN-858] Generate patch to each Spark 3.x minor version

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

Add the following patch files in directory `incubator-celeborn/tree/spark3-patch/assets/spark-patch` :

1. Celeborn_Dynamic_Allocation_spark3_0.patch
2. Celeborn_Dynamic_Allocation_spark3_1.patch
3. Celeborn_Dynamic_Allocation_spark3_2.patch
4. Celeborn_Dynamic_Allocation_spark3_3.patch

Delete a patch at the same time：

1. Celeborn_Dynamic_Allocation_spark3.patch

Modified `Support Spark Dynamic Allocation` in incubator-celeborn/README.md ：

![image](https://github.com/apache/incubator-celeborn/assets/108530647/61e2e69b-d3f5-4d11-a20b-374622936443)

### Why are the changes needed?

Convenient for customers to apply patches in Spark 3.X for `Support Spark Dynamic Allocation`

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

no

### How was this patch tested?

yes. All patch files can be applied to the corresponding version of spark source code through `git apply`  without any code conflicts.

Closes #2085 from lukeyan2023/spark3-patch.

Authored-by: Luke Yan <108530647+lukeyan2023@users.noreply.github.com>
Signed-off-by: zky.zhoukeyong <zky.zhoukeyong@alibaba-inc.com>
This commit is contained in:
Luke Yan 2023-11-10 15:35:54 +08:00 committed by zky.zhoukeyong
parent eb1be3fbf8
commit c7c2f6a35a
5 changed files with 270 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@ -333,7 +333,10 @@ for more details.
For Spark versions < 3.5.0, we provide a patch to enable users to use Spark with DRA and Celeborn.
- For Spark 2.x check [Spark2 Patch](assets/spark-patch/Celeborn_Dynamic_Allocation_spark2.patch).
- For Spark 3.0-3.3 check [Spark3 Patch](assets/spark-patch/Celeborn_Dynamic_Allocation_spark3.patch).
- For Spark 3.0 check [Spark3.0 Patch](assets/spark-patch/Celeborn_Dynamic_Allocation_spark3_0.patch).
- For Spark 3.1 check [Spark3.1 Patch](assets/spark-patch/Celeborn_Dynamic_Allocation_spark3_1.patch).
- For Spark 3.2 check [Spark3.2 Patch](assets/spark-patch/Celeborn_Dynamic_Allocation_spark3_2.patch).
- For Spark 3.3 check [Spark3.3 Patch](assets/spark-patch/Celeborn_Dynamic_Allocation_spark3_3.patch).
- For Spark 3.4 check [Spark3.4 Patch](assets/spark-patch/Celeborn_Dynamic_Allocation_spark3_4.patch).
### Metrics

87
assets/spark-patch/Celeborn_Dynamic_Allocation_spark3_0.patch Normal file
View File

@ -0,0 +1,87 @@
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
Subject: [PATCH] [CORE][SHUFFLE] Support enabling DRA with Apache Celeborn
---
Index: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala (revision 65ac1e75dc468f53fc778cd2ce1ba3f21067aab8)
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala (revision 1f5cb2ec9e9652d03c9775954b69a708b5d05ab3)
@@ -198,7 +198,7 @@
if (!conf.get(config.SHUFFLE_SERVICE_ENABLED)) {
if (conf.get(config.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED)) {
logWarning("Dynamic allocation without a shuffle service is an experimental feature.")
- } else if (!testing) {
+ } else if (!testing && !Utils.isCelebornEnabled(conf)) {
throw new SparkException("Dynamic allocation of executors requires the external " +
"shuffle service. You may enable this through spark.shuffle.service.enabled.")
}
Index: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala (revision 65ac1e75dc468f53fc778cd2ce1ba3f21067aab8)
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala (revision 1f5cb2ec9e9652d03c9775954b69a708b5d05ab3)
@@ -1851,7 +1851,8 @@
// if the cluster manager explicitly tells us that the entire worker was lost, then
// we know to unregister shuffle output. (Note that "worker" specifically refers to the process
// from a Standalone cluster, where the shuffle service lives in the Worker.)
- val fileLost = workerLost || !env.blockManager.externalShuffleServiceEnabled
+ val fileLost = !Utils.isCelebornEnabled(sc.getConf) &&
+ (workerLost || !env.blockManager.externalShuffleServiceEnabled)
removeExecutorAndUnregisterOutputs(
execId = execId,
fileLost = fileLost,
Index: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala (revision 65ac1e75dc468f53fc778cd2ce1ba3f21067aab8)
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala (revision 1f5cb2ec9e9652d03c9775954b69a708b5d05ab3)
@@ -944,7 +944,7 @@
// The reason is the next stage wouldn't be able to fetch the data from this dead executor
// so we would need to rerun these tasks on other executors.
if (tasks(0).isInstanceOf[ShuffleMapTask] && !env.blockManager.externalShuffleServiceEnabled
- && !isZombie) {
+ && !isZombie && !Utils.isCelebornEnabled(conf)) {
for ((tid, info) <- taskInfos if info.executorId == execId) {
val index = taskInfos(tid).index
// We may have a running task whose partition has been marked as successful,
Index: core/src/main/scala/org/apache/spark/util/Utils.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala (revision 65ac1e75dc468f53fc778cd2ce1ba3f21067aab8)
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala (revision 1f5cb2ec9e9652d03c9775954b69a708b5d05ab3)
@@ -2934,6 +2934,9 @@
props.forEach((k, v) => resultProps.put(k, v))
resultProps
}
+
+ def isCelebornEnabled(conf: SparkConf): Boolean =
+ conf.get("spark.shuffle.manager", "sort").contains("celeborn")
}
private[util] object CallerContext extends Logging {

91
assets/spark-patch/Celeborn_Dynamic_Allocation_spark3_1.patch Normal file
View File

@ -0,0 +1,91 @@
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
Subject: [PATCH] [CORE][SHUFFLE] Support enabling DRA with Apache Celeborn
---
Index: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala (revision d1f8a503a26bcfb4e466d9accc5fa241a7933667)
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala (revision c8609fa08534e1cc6fd377b3573e5e65ddc79f15)
@@ -210,7 +210,7 @@
(decommissionEnabled &&
conf.get(config.STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED))) {
logWarning("Dynamic allocation without a shuffle service is an experimental feature.")
- } else if (!testing) {
+ } else if (!testing && !Utils.isCelebornEnabled(conf)) {
throw new SparkException("Dynamic allocation of executors requires the external " +
"shuffle service. You may enable this through spark.shuffle.service.enabled.")
}
Index: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala (revision d1f8a503a26bcfb4e466d9accc5fa241a7933667)
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala (revision c8609fa08534e1cc6fd377b3573e5e65ddc79f15)
@@ -2080,7 +2080,8 @@
// if the cluster manager explicitly tells us that the entire worker was lost, then
// we know to unregister shuffle output. (Note that "worker" specifically refers to the process
// from a Standalone cluster, where the shuffle service lives in the Worker.)
- val fileLost = workerHost.isDefined || !env.blockManager.externalShuffleServiceEnabled
+ val fileLost = !Utils.isCelebornEnabled(sc.getConf) &&
+ (workerHost.isDefined || !env.blockManager.externalShuffleServiceEnabled)
removeExecutorAndUnregisterOutputs(
execId = execId,
fileLost = fileLost,
Index: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala (revision d1f8a503a26bcfb4e466d9accc5fa241a7933667)
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala (revision c8609fa08534e1cc6fd377b3573e5e65ddc79f15)
@@ -973,7 +973,8 @@
// and we are not using an external shuffle server which could serve the shuffle outputs.
// The reason is the next stage wouldn't be able to fetch the data from this dead executor
// so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie &&
+ !Utils.isCelebornEnabled(conf)) {
for ((tid, info) <- taskInfos if info.executorId == execId) {
val index = taskInfos(tid).index
// We may have a running task whose partition has been marked as successful,
Index: core/src/main/scala/org/apache/spark/util/Utils.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala (revision d1f8a503a26bcfb4e466d9accc5fa241a7933667)
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala (revision c8609fa08534e1cc6fd377b3573e5e65ddc79f15)
@@ -3057,7 +3057,12 @@
0
}
}
+
+ def isCelebornEnabled(conf: SparkConf): Boolean =
+ conf.get("spark.shuffle.manager", "sort").contains("celeborn")
}
+
+
private[util] object CallerContext extends Logging {
val callerContextSupported: Boolean = {

0
assets/spark-patch/Celeborn_Dynamic_Allocation_spark3.patch → assets/spark-patch/Celeborn_Dynamic_Allocation_spark3_2.patch
View File

88
assets/spark-patch/Celeborn_Dynamic_Allocation_spark3_3.patch Normal file
View File

@ -0,0 +1,88 @@
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
Subject: [PATCH] [CORE][SHUFFLE] Support enabling DRA with Apache Celeborn
---
Index: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala (revision 8c2b3319c6734250ff9d72f3d7e5cab56b142195)
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala (revision 5e318f8fbeadec03986b3ad9cad4c1cda33c9ed2)
@@ -209,7 +209,7 @@
} else if (decommissionEnabled &&
conf.get(config.STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED)) {
logInfo("Shuffle data decommission is enabled without a shuffle service.")
- } else if (!testing) {
+ } else if (!testing && !Utils.isCelebornEnabled(conf)) {
throw new SparkException("Dynamic allocation of executors requires the external " +
"shuffle service. You may enable this through spark.shuffle.service.enabled.")
}
Index: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala (revision 8c2b3319c6734250ff9d72f3d7e5cab56b142195)
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala (revision 5e318f8fbeadec03986b3ad9cad4c1cda33c9ed2)
@@ -2414,7 +2414,8 @@
// if the cluster manager explicitly tells us that the entire worker was lost, then
// we know to unregister shuffle output. (Note that "worker" specifically refers to the process
// from a Standalone cluster, where the shuffle service lives in the Worker.)
- val fileLost = workerHost.isDefined || !env.blockManager.externalShuffleServiceEnabled
+ val fileLost = !Utils.isCelebornEnabled(sc.getConf) &&
+ (workerHost.isDefined || !env.blockManager.externalShuffleServiceEnabled)
removeExecutorAndUnregisterOutputs(
execId = execId,
fileLost = fileLost,
Index: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala (revision 8c2b3319c6734250ff9d72f3d7e5cab56b142195)
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala (revision 5e318f8fbeadec03986b3ad9cad4c1cda33c9ed2)
@@ -1032,7 +1032,8 @@
// and we are not using an external shuffle server which could serve the shuffle outputs.
// The reason is the next stage wouldn't be able to fetch the data from this dead executor
// so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie &&
+ !Utils.isCelebornEnabled(conf)) {
for ((tid, info) <- taskInfos if info.executorId == execId) {
val index = info.index
// We may have a running task whose partition has been marked as successful,
Index: core/src/main/scala/org/apache/spark/util/Utils.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala (revision 8c2b3319c6734250ff9d72f3d7e5cab56b142195)
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala (revision 5e318f8fbeadec03986b3ad9cad4c1cda33c9ed2)
@@ -3246,6 +3246,9 @@
files.toSeq
}
+ def isCelebornEnabled(conf: SparkConf): Boolean =
+ conf.get("spark.shuffle.manager", "sort").contains("celeborn")
+
/**
* Return the median number of a long array
*