Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[#1881] fix(client-spark): Make spark client uniifle class under org.apache.uniffle package to avoid conflict #1882

Draft
wants to merge 8 commits into
base: master
Choose a base branch
from
Draft

[#1881] fix(client-spark): Make spark client uniifle class under org.apache.uniffle package to avoid conflict #1882

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
005403d
Make spark client uniifle class under org.apache.uniffle package to a…
maobaolong Jul 10, 2024
0752044
Fix build error
maobaolong Jul 10, 2024
f3d8131
Fix build error
maobaolong Jul 10, 2024
7eccbf0
Fix build error
maobaolong Jul 10, 2024
5c3d5db
Fix build error
maobaolong Jul 10, 2024
1feff30
Fix style
maobaolong Jul 10, 2024
0ee9803
Fix build error
maobaolong Jul 10, 2024
28fe7d7
Fix docker rename
maobaolong Jul 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/docker.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,7 +123,7 @@ jobs:
docker exec rss-spark-master-1 /bin/bash -c "cat /example.scala | /opt/spark/bin/spark-shell \
--master spark://rss-spark-master-1:7077 \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
--conf spark.shuffle.manager=org.apache.spark.shuffle.RssShuffleManager \
--conf spark.shuffle.manager=org.apache.uniffle.spark.shuffle.RssShuffleManager \
--conf spark.rss.coordinator.quorum=rss-coordinator-1:19999,rss-coordinator-2:19999 \
--conf spark.rss.storage.type=MEMORY_LOCALFILE \
--conf spark.task.maxFailures=4 \
Expand Down
4 changes: 2 additions & 2 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,7 @@ Deploy Steps:
# Uniffle transmits serialized shuffle data over network, therefore a serializer that supports relocation of
# serialized object should be used.
spark.serializer org.apache.spark.serializer.KryoSerializer # this could also be in the spark-defaults.conf
spark.shuffle.manager org.apache.spark.shuffle.RssShuffleManager
spark.shuffle.manager org.apache.uniffle.spark.shuffle.RssShuffleManager
spark.rss.coordinator.quorum <coordinatorIp1>:19999,<coordinatorIp2>:19999
# Note: For Spark2, spark.sql.adaptive.enabled should be false because Spark2 doesn't support AQE.
```
Expand All @@ -269,7 +269,7 @@ After apply the patch and rebuild spark, add following configuration in spark co
```
For spark3.5 or above just add one more configuration:
```
spark.shuffle.sort.io.plugin.class org.apache.spark.shuffle.RssShuffleDataIo
spark.shuffle.sort.io.plugin.class org.apache.uniffle.spark.shuffle.RssShuffleDataIo
```

### Deploy MapReduce Client
Expand Down
22 changes: 11 additions & 11 deletions ...-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/RssShuffleManagerBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,18 +44,7 @@
import org.apache.spark.SparkConf;
import org.apache.spark.SparkEnv;
import org.apache.spark.SparkException;
import org.apache.spark.shuffle.RssShuffleHandle;
import org.apache.spark.shuffle.RssSparkConfig;
import org.apache.spark.shuffle.RssSparkShuffleUtils;
import org.apache.spark.shuffle.RssStageInfo;
import org.apache.spark.shuffle.RssStageResubmitManager;
import org.apache.spark.shuffle.ShuffleHandleInfoManager;
import org.apache.spark.shuffle.ShuffleManager;
import org.apache.spark.shuffle.SparkVersionUtils;
import org.apache.spark.shuffle.handle.MutableShuffleHandleInfo;
import org.apache.spark.shuffle.handle.ShuffleHandleInfo;
import org.apache.spark.shuffle.handle.SimpleShuffleHandleInfo;
import org.apache.spark.shuffle.handle.StageAttemptShuffleHandleInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -84,6 +73,17 @@
import org.apache.uniffle.common.rpc.StatusCode;
import org.apache.uniffle.common.util.RetryUtils;
import org.apache.uniffle.shuffle.BlockIdManager;
import org.apache.uniffle.spark.shuffle.RssShuffleHandle;
import org.apache.uniffle.spark.shuffle.RssSparkConfig;
import org.apache.uniffle.spark.shuffle.RssSparkShuffleUtils;
import org.apache.uniffle.spark.shuffle.RssStageInfo;
import org.apache.uniffle.spark.shuffle.RssStageResubmitManager;
import org.apache.uniffle.spark.shuffle.ShuffleHandleInfoManager;
import org.apache.uniffle.spark.shuffle.SparkVersionUtils;
import org.apache.uniffle.spark.shuffle.handle.MutableShuffleHandleInfo;
import org.apache.uniffle.spark.shuffle.handle.ShuffleHandleInfo;
import org.apache.uniffle.spark.shuffle.handle.SimpleShuffleHandleInfo;
import org.apache.uniffle.spark.shuffle.handle.StageAttemptShuffleHandleInfo;

import static org.apache.uniffle.common.config.RssClientConf.HADOOP_CONFIG_KEY_PREFIX;
import static org.apache.uniffle.common.config.RssClientConf.RSS_CLIENT_REMOTE_STORAGE_USE_LOCAL_CONF_ENABLED;
Expand Down
4 changes: 2 additions & 2 deletions ...k/common/src/main/java/org/apache/uniffle/shuffle/manager/RssShuffleManagerInterface.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,11 @@
import java.util.Map;

import org.apache.spark.SparkException;
import org.apache.spark.shuffle.handle.MutableShuffleHandleInfo;
import org.apache.spark.shuffle.handle.ShuffleHandleInfo;

import org.apache.uniffle.common.ReceivingFailureServer;
import org.apache.uniffle.shuffle.BlockIdManager;
import org.apache.uniffle.spark.shuffle.handle.MutableShuffleHandleInfo;
import org.apache.uniffle.spark.shuffle.handle.ShuffleHandleInfo;

/**
* This is a proxy interface that mainly delegates the un-registration of shuffles to the
Expand Down
4 changes: 2 additions & 2 deletions ...rk/common/src/main/java/org/apache/uniffle/shuffle/manager/ShuffleManagerGrpcService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,6 @@

import com.google.protobuf.UnsafeByteOperations;
import io.grpc.stub.StreamObserver;
import org.apache.spark.shuffle.handle.MutableShuffleHandleInfo;
import org.apache.spark.shuffle.handle.StageAttemptShuffleHandleInfo;
import org.roaringbitmap.longlong.Roaring64NavigableMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -42,6 +40,8 @@
import org.apache.uniffle.proto.RssProtos;
import org.apache.uniffle.proto.ShuffleManagerGrpc.ShuffleManagerImplBase;
import org.apache.uniffle.shuffle.BlockIdManager;
import org.apache.uniffle.spark.shuffle.handle.MutableShuffleHandleInfo;
import org.apache.uniffle.spark.shuffle.handle.StageAttemptShuffleHandleInfo;

public class ShuffleManagerGrpcService extends ShuffleManagerImplBase {
private static final Logger LOG = LoggerFactory.getLogger(ShuffleManagerGrpcService.class);
Expand Down
5 changes: 3 additions & 2 deletions ...pache/spark/shuffle/RssShuffleHandle.java → ...iffle/spark/shuffle/RssShuffleHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,17 +15,18 @@
* limitations under the License.
*/

package org.apache.spark.shuffle;
package org.apache.uniffle.spark.shuffle;

import java.util.List;
import java.util.Map;

import org.apache.spark.ShuffleDependency;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.shuffle.handle.SimpleShuffleHandleInfo;
import org.apache.spark.shuffle.ShuffleHandle;

import org.apache.uniffle.common.RemoteStorageInfo;
import org.apache.uniffle.common.ShuffleServerInfo;
import org.apache.uniffle.spark.shuffle.handle.SimpleShuffleHandleInfo;

public class RssShuffleHandle<K, V, C> extends ShuffleHandle {

Expand Down
2 changes: 1 addition & 1 deletion .../apache/spark/shuffle/RssSparkConfig.java → ...uniffle/spark/shuffle/RssSparkConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle;
package org.apache.uniffle.spark.shuffle;

import java.util.Set;

Expand Down
8 changes: 5 additions & 3 deletions ...e/spark/shuffle/RssSparkShuffleUtils.java → ...e/spark/shuffle/RssSparkShuffleUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle;
package org.apache.uniffle.spark.shuffle;

import java.io.IOException;
import java.lang.reflect.Constructor;
Expand All @@ -35,7 +35,8 @@
import org.apache.spark.SparkContext;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.deploy.SparkHadoopUtil;
import org.apache.spark.shuffle.handle.SimpleShuffleHandleInfo;
import org.apache.spark.shuffle.FetchFailedException;
import org.apache.spark.shuffle.ShuffleManager;
import org.apache.spark.storage.BlockManagerId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -55,9 +56,10 @@
import org.apache.uniffle.common.exception.RssException;
import org.apache.uniffle.common.exception.RssFetchFailedException;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.spark.shuffle.handle.SimpleShuffleHandleInfo;

import static org.apache.spark.shuffle.RssSparkConfig.RSS_RESUBMIT_STAGE_WITH_FETCH_FAILURE_ENABLED;
import static org.apache.uniffle.common.util.Constants.DRIVER_HOST;
import static org.apache.uniffle.spark.shuffle.RssSparkConfig.RSS_RESUBMIT_STAGE_WITH_FETCH_FAILURE_ENABLED;

public class RssSparkShuffleUtils {

Expand Down
2 changes: 1 addition & 1 deletion ...rg/apache/spark/shuffle/RssStageInfo.java → ...e/uniffle/spark/shuffle/RssStageInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle;
package org.apache.uniffle.spark.shuffle;

public class RssStageInfo {
private String stageAttemptIdAndNumber;
Expand Down
2 changes: 1 addition & 1 deletion ...park/shuffle/RssStageResubmitManager.java → ...park/shuffle/RssStageResubmitManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle;
package org.apache.uniffle.spark.shuffle;

import java.util.Map;
import java.util.Set;
Expand Down
5 changes: 2 additions & 3 deletions ...ark/shuffle/ShuffleHandleInfoManager.java → ...ark/shuffle/ShuffleHandleInfoManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,15 +15,14 @@
* limitations under the License.
*/

package org.apache.spark.shuffle;
package org.apache.uniffle.spark.shuffle;

import java.io.Closeable;
import java.io.IOException;
import java.util.Map;

import org.apache.spark.shuffle.handle.ShuffleHandleInfo;

import org.apache.uniffle.common.util.JavaUtils;
import org.apache.uniffle.spark.shuffle.handle.ShuffleHandleInfo;

public class ShuffleHandleInfoManager implements Closeable {
private Map<Integer, ShuffleHandleInfo> shuffleIdToShuffleHandleInfo;
Expand Down
2 changes: 1 addition & 1 deletion ...ache/spark/shuffle/SparkVersionUtils.java → ...ffle/spark/shuffle/SparkVersionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle;
package org.apache.uniffle.spark.shuffle;

import org.apache.spark.package$;
import org.apache.spark.util.VersionUtils;
Expand Down
2 changes: 1 addition & 1 deletion ...ffle/handle/MutableShuffleHandleInfo.java → ...ffle/handle/MutableShuffleHandleInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.handle;
package org.apache.uniffle.spark.shuffle.handle;

import java.util.ArrayList;
import java.util.Collections;
Expand Down
2 changes: 1 addition & 1 deletion ...ark/shuffle/handle/ShuffleHandleInfo.java → ...ark/shuffle/handle/ShuffleHandleInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.handle;
package org.apache.uniffle.spark.shuffle.handle;

import java.util.List;
import java.util.Map;
Expand Down
2 changes: 1 addition & 1 deletion ...shuffle/handle/ShuffleHandleInfoBase.java → ...shuffle/handle/ShuffleHandleInfoBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.handle;
package org.apache.uniffle.spark.shuffle.handle;

import java.io.Serializable;

Expand Down
2 changes: 1 addition & 1 deletion ...uffle/handle/SimpleShuffleHandleInfo.java → ...uffle/handle/SimpleShuffleHandleInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.handle;
package org.apache.uniffle.spark.shuffle.handle;

import java.io.Serializable;
import java.util.List;
Expand Down
2 changes: 1 addition & 1 deletion ...handle/StageAttemptShuffleHandleInfo.java → ...handle/StageAttemptShuffleHandleInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.handle;
package org.apache.uniffle.spark.shuffle.handle;

import java.util.LinkedList;
import java.util.List;
Expand Down
4 changes: 2 additions & 2 deletions ...huffle/reader/RssFetchFailedIterator.java → ...huffle/reader/RssFetchFailedIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.reader;
package org.apache.uniffle.spark.shuffle.reader;

import java.io.IOException;
import java.util.Objects;
Expand All @@ -25,7 +25,6 @@
import scala.collection.Iterator;

import org.apache.spark.shuffle.FetchFailedException;
import org.apache.spark.shuffle.RssSparkShuffleUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -36,6 +35,7 @@
import org.apache.uniffle.common.ClientType;
import org.apache.uniffle.common.exception.RssException;
import org.apache.uniffle.common.exception.RssFetchFailedException;
import org.apache.uniffle.spark.shuffle.RssSparkShuffleUtils;

public class RssFetchFailedIterator<K, C> extends AbstractIterator<Product2<K, C>> {
private static final Logger LOG = LoggerFactory.getLogger(RssFetchFailedIterator.class);
Expand Down
4 changes: 2 additions & 2 deletions ...huffle/reader/RssShuffleDataIterator.java → ...huffle/reader/RssShuffleDataIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.reader;
package org.apache.uniffle.spark.shuffle.reader;

import java.io.IOException;
import java.nio.ByteBuffer;
Expand All @@ -33,7 +33,6 @@
import org.apache.spark.serializer.DeserializationStream;
import org.apache.spark.serializer.Serializer;
import org.apache.spark.serializer.SerializerInstance;
import org.apache.spark.shuffle.RssSparkConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -42,6 +41,7 @@
import org.apache.uniffle.common.compression.Codec;
import org.apache.uniffle.common.config.RssConf;
import org.apache.uniffle.common.util.RssUtils;
import org.apache.uniffle.spark.shuffle.RssSparkConfig;

public class RssShuffleDataIterator<K, C> extends AbstractIterator<Product2<K, C>> {

Expand Down
2 changes: 1 addition & 1 deletion ...e/spark/shuffle/writer/AddBlockEvent.java → ...e/spark/shuffle/writer/AddBlockEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.writer;
package org.apache.uniffle.spark.shuffle.writer;

import java.util.ArrayList;
import java.util.List;
Expand Down
2 changes: 1 addition & 1 deletion .../shuffle/writer/BlockFailureCallback.java → .../shuffle/writer/BlockFailureCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.writer;
package org.apache.uniffle.spark.shuffle.writer;

import org.apache.uniffle.common.ShuffleBlockInfo;

Expand Down
2 changes: 1 addition & 1 deletion .../shuffle/writer/BlockSuccessCallback.java → .../shuffle/writer/BlockSuccessCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.writer;
package org.apache.uniffle.spark.shuffle.writer;

import org.apache.uniffle.common.ShuffleBlockInfo;

Expand Down
4 changes: 2 additions & 2 deletions .../shuffle/writer/BufferManagerOptions.java → .../shuffle/writer/BufferManagerOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,14 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.writer;
package org.apache.uniffle.spark.shuffle.writer;

import org.apache.spark.SparkConf;
import org.apache.spark.shuffle.RssSparkConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.uniffle.common.exception.RssException;
import org.apache.uniffle.spark.shuffle.RssSparkConfig;

public class BufferManagerOptions {

Expand Down
2 changes: 1 addition & 1 deletion ...ache/spark/shuffle/writer/DataPusher.java → ...ffle/spark/shuffle/writer/DataPusher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.writer;
package org.apache.uniffle.spark.shuffle.writer;

import java.io.Closeable;
import java.io.IOException;
Expand Down
5 changes: 2 additions & 3 deletions ...shuffle/writer/TaskAttemptAssignment.java → ...shuffle/writer/TaskAttemptAssignment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,15 +15,14 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.writer;
package org.apache.uniffle.spark.shuffle.writer;

import java.util.List;
import java.util.Map;

import org.apache.spark.shuffle.handle.ShuffleHandleInfo;

import org.apache.uniffle.common.ShuffleServerInfo;
import org.apache.uniffle.common.exception.RssException;
import org.apache.uniffle.spark.shuffle.handle.ShuffleHandleInfo;

/** This class is to get the partition assignment for ShuffleWriter. */
public class TaskAttemptAssignment {
Expand Down
2 changes: 1 addition & 1 deletion .../writer/WrappedByteArrayOutputStream.java → .../writer/WrappedByteArrayOutputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.shuffle.writer;
package org.apache.uniffle.spark.shuffle.writer;

import java.io.ByteArrayOutputStream;

Expand Down
Loading
Loading