You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/05/25 02:29:22 UTC
[incubator-celeborn] branch branch-0.2 updated: [CELEBORN-604][SPARK] Support Spark 3.4 (#1515)
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch branch-0.2
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/branch-0.2 by this push:
new aebd2ef2c [CELEBORN-604][SPARK] Support Spark 3.4 (#1515)
aebd2ef2c is described below
commit aebd2ef2c7b992371498d8c5d0c25752b9037e0b
Author: Keyong Zhou <zh...@apache.org>
AuthorDate: Thu May 25 10:29:19 2023 +0800
[CELEBORN-604][SPARK] Support Spark 3.4 (#1515)
---
.github/workflows/maven.yml | 1 +
README.md | 2 +-
.../shuffle/celeborn/HashBasedShuffleWriter.java | 10 +++++++---
.../shuffle/celeborn/SortBasedShuffleWriter.java | 10 +++++++---
.../shuffle/celeborn/RssShuffleWriterSuiteJ.java | 15 ++++++++++++---
pom.xml | 19 +++++++++++++++++++
tests/spark-it/pom.xml | 11 +++++++++++
7 files changed, 58 insertions(+), 10 deletions(-)
diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
index 492f9aedf..3b195106a 100644
--- a/.github/workflows/maven.yml
+++ b/.github/workflows/maven.yml
@@ -56,6 +56,7 @@ jobs:
- '3.1'
- '3.2'
- '3.3'
+ - '3.4'
steps:
- uses: actions/checkout@v2
- name: Setup JDK ${{ matrix.java }}
diff --git a/README.md b/README.md
index 9727a2f0c..752023491 100644
--- a/README.md
+++ b/README.md
@@ -45,7 +45,7 @@ Celeborn supports Spark 2.4/3.0/3.1/3.2/3.3 and only tested under Java 8.
Build for Spark
```
-./build/make-distribution.sh -Pspark-2.4/-Pspark-3.0/-Pspark-3.1/-Pspark-3.2/-Pspark-3.3
+./build/make-distribution.sh -Pspark-2.4/-Pspark-3.0/-Pspark-3.1/-Pspark-3.2/-Pspark-3.3/-Pspark-3.4
```
package apache-celeborn-${project.version}-bin.tgz will be generated.
diff --git a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java
index e54ba063e..ec3590953 100644
--- a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java
+++ b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java
@@ -39,7 +39,6 @@ import org.apache.spark.serializer.SerializerInstance;
import org.apache.spark.shuffle.ShuffleWriteMetricsReporter;
import org.apache.spark.shuffle.ShuffleWriter;
import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
-import org.apache.spark.sql.execution.PartitionIdPassthrough;
import org.apache.spark.sql.execution.UnsafeRowSerializer;
import org.apache.spark.sql.execution.columnar.RssBatchBuilder;
import org.apache.spark.sql.execution.columnar.RssColumnarBatchBuilder;
@@ -191,8 +190,13 @@ public class HashBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
@VisibleForTesting
boolean canUseFastWrite() {
- return dep.serializer() instanceof UnsafeRowSerializer
- && partitioner instanceof PartitionIdPassthrough;
+ boolean keyIsPartitionId = false;
+ if (dep.serializer() instanceof UnsafeRowSerializer) {
+ // SPARK-39391 renames PartitionIdPassthrough's package
+ String partitionerClassName = partitioner.getClass().getSimpleName();
+ keyIsPartitionId = "PartitionIdPassthrough".equals(partitionerClassName);
+ }
+ return keyIsPartitionId;
}
private void fastColumnarWrite0(scala.collection.Iterator iterator) throws IOException {
diff --git a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
index ce9c1f85d..fc9223709 100644
--- a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
+++ b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
@@ -39,7 +39,6 @@ import org.apache.spark.serializer.SerializerInstance;
import org.apache.spark.shuffle.ShuffleWriteMetricsReporter;
import org.apache.spark.shuffle.ShuffleWriter;
import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
-import org.apache.spark.sql.execution.PartitionIdPassthrough;
import org.apache.spark.sql.execution.UnsafeRowSerializer;
import org.apache.spark.sql.execution.metric.SQLMetric;
import org.apache.spark.storage.BlockManagerId;
@@ -157,8 +156,13 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
@VisibleForTesting
boolean canUseFastWrite() {
- return dep.serializer() instanceof UnsafeRowSerializer
- && partitioner instanceof PartitionIdPassthrough;
+ boolean keyIsPartitionId = false;
+ if (dep.serializer() instanceof UnsafeRowSerializer) {
+ // SPARK-39391 renames PartitionIdPassthrough's package
+ String partitionerClassName = partitioner.getClass().getSimpleName();
+ keyIsPartitionId = "PartitionIdPassthrough".equals(partitionerClassName);
+ }
+ return keyIsPartitionId;
}
private void fastWrite0(scala.collection.Iterator iterator) throws IOException {
diff --git a/client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/RssShuffleWriterSuiteJ.java b/client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/RssShuffleWriterSuiteJ.java
index c7e59ce97..5c600e553 100644
--- a/client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/RssShuffleWriterSuiteJ.java
+++ b/client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/RssShuffleWriterSuiteJ.java
@@ -51,7 +51,6 @@ import org.apache.spark.serializer.Serializer;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.catalyst.expressions.UnsafeProjection;
import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
-import org.apache.spark.sql.execution.PartitionIdPassthrough;
import org.apache.spark.sql.execution.UnsafeRowSerializer;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.IntegerType$;
@@ -76,6 +75,7 @@ import org.apache.celeborn.common.CelebornConf;
import org.apache.celeborn.common.identity.UserIdentifier;
import org.apache.celeborn.common.network.util.JavaUtils;
import org.apache.celeborn.common.util.Utils;
+import org.apache.celeborn.reflect.DynConstructors;
public class RssShuffleWriterSuiteJ {
@@ -93,7 +93,7 @@ public class RssShuffleWriterSuiteJ {
private final UserIdentifier userIdentifier = new UserIdentifier("mock", "mock");
private final int numMaps = 10;
- private final int numPartitions = 10;
+ private final Integer numPartitions = 10;
private final SparkConf sparkConf = new SparkConf(false);
private final BlockManagerId bmId = BlockManagerId.apply("execId", "host", 1, None$.empty());
@@ -202,8 +202,17 @@ public class RssShuffleWriterSuiteJ {
throws Exception {
final boolean useUnsafe = serializer instanceof UnsafeRowSerializer;
+ DynConstructors.Ctor<Partitioner> partitionIdPassthroughCtor =
+ DynConstructors.builder()
+ // for Spark 3.3 and previous
+ .impl("org.apache.spark.sql.execution.PartitionIdPassthrough", int.class)
+ // for Spark 3.4
+ .impl("org.apache.spark.PartitionIdPassthrough", int.class)
+ .build();
final Partitioner partitioner =
- useUnsafe ? new PartitionIdPassthrough(numPartitions) : new HashPartitioner(numPartitions);
+ useUnsafe
+ ? partitionIdPassthroughCtor.newInstance(numPartitions)
+ : new HashPartitioner(numPartitions);
Mockito.doReturn(partitioner).when(dependency).partitioner();
Mockito.doReturn(serializer).when(dependency).serializer();
diff --git a/pom.xml b/pom.xml
index a6500cbf2..79c4fac30 100644
--- a/pom.xml
+++ b/pom.xml
@@ -960,6 +960,25 @@
</properties>
</profile>
+ <profile>
+ <id>spark-3.4</id>
+ <modules>
+ <module>client-spark/common</module>
+ <module>client-spark/spark-3</module>
+ <module>client-spark/spark-3-shaded</module>
+ <module>tests/spark-it</module>
+ </modules>
+ <properties>
+ <jackson.version>2.14.2</jackson.version>
+ <jackson.databind.version>2.14.2</jackson.databind.version>
+ <lz4-java.version>1.8.0</lz4-java.version>
+ <scala.version>2.12.17</scala.version>
+ <scala.binary.version>2.12</scala.binary.version>
+ <spark.version>3.4.0</spark.version>
+ <zstd-jni.version>1.5.2-5</zstd-jni.version>
+ </properties>
+ </profile>
+
<profile>
<id>jdk-8</id>
<activation>
diff --git a/tests/spark-it/pom.xml b/tests/spark-it/pom.xml
index a608b9dd8..e9242c20e 100644
--- a/tests/spark-it/pom.xml
+++ b/tests/spark-it/pom.xml
@@ -136,5 +136,16 @@
</dependency>
</dependencies>
</profile>
+ <profile>
+ <id>spark-3.4</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-client-spark-3_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
</profiles>
</project>