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

Jump to bottom

[#1373][FOLLOWUP] fix(spark): shuffle manager rpc service invalid when partition data reassign is enabled #1583

Merged
merged 6 commits into from
Apr 1, 2024
Merged

[#1373][FOLLOWUP] fix(spark): shuffle manager rpc service invalid when partition data reassign is enabled #1583

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
81e129c
[#1373][followup] fix(client):Fix task fail retry parameter not work
dingshun3016 Mar 14, 2024
c0a09ba
fix code style
dingshun3016 Mar 15, 2024
d795246
fix the spark2 and rename
dingshun3016 Mar 16, 2024
f28ca7d
refact code structure
Apr 1, 2024
c79ef03
fix config
Apr 1, 2024
f488c59
fix code style
Apr 1, 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
14 changes: 10 additions & 4 deletions client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,7 @@ public class RssShuffleManager extends RssShuffleManagerBase {
private Map<Integer, ShuffleHandleInfo> shuffleIdToShuffleHandleInfo;
/** Whether to enable the dynamic shuffleServer function rewrite and reread functions */
private boolean rssResubmitStage;
private boolean taskFailRetry;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

taskFailRetry -> taskReattemptOnFailures

Is is better?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think this is only valid for task write failure, in the future, it could be applied on the not-enough-capacity disk servers

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

taskFailureRetryEnabled

/** A list of shuffleServer for Write failures */
private Set<String> failuresShuffleServerIds;
/**
Expand Down Expand Up @@ -239,10 +240,15 @@ public RssShuffleManager(SparkConf conf, boolean isDriver) {
this.rssResubmitStage =
rssConf.getBoolean(RssClientConfig.RSS_RESUBMIT_STAGE, false)
&& RssSparkShuffleUtils.isStageResubmitSupported();
this.taskFailRetry =
sparkConf.getBoolean(
RssSparkConfig.SPARK_RSS_CONFIG_PREFIX
+ RssClientConf.RSS_TASK_FAILED_RETRY_ENABLED.key(),
RssClientConf.RSS_TASK_FAILED_RETRY_ENABLED.defaultValue());
if (isDriver) {
heartBeatScheduledExecutorService =
ThreadUtils.getDaemonSingleThreadScheduledExecutor("rss-heartbeat");
if (rssResubmitStage) {
if (rssResubmitStage || taskFailRetry) {
LOG.info("stage resubmit is supported and enabled");
// start shuffle manager server
rssConf.set(RPC_SERVER_PORT, 0);
Expand Down Expand Up @@ -471,7 +477,7 @@ public <K, V, C> ShuffleHandle registerShuffle(

shuffleIdToPartitionNum.putIfAbsent(shuffleId, dependency.partitioner().numPartitions());
shuffleIdToNumMapTasks.putIfAbsent(shuffleId, dependency.rdd().partitions().length);
if (rssResubmitStage) {
if (rssResubmitStage || taskFailRetry) {
ShuffleHandleInfo handleInfo =
new ShuffleHandleInfo(shuffleId, partitionToServers, remoteStorage);
shuffleIdToShuffleHandleInfo.put(shuffleId, handleInfo);
Expand Down Expand Up @@ -509,7 +515,7 @@ public <K, V> ShuffleWriter<K, V> getWriter(
writeMetrics = context.taskMetrics().shuffleWriteMetrics();
}
ShuffleHandleInfo shuffleHandleInfo;
if (rssResubmitStage) {
if (rssResubmitStage || taskFailRetry) {
// Get the ShuffleServer list from the Driver based on the shuffleId
shuffleHandleInfo = getRemoteShuffleHandleInfo(shuffleId);
} else {
Expand Down Expand Up @@ -651,7 +657,7 @@ public <K, C> ShuffleReader<K, C> getReaderImpl(
final int partitionNum = rssShuffleHandle.getDependency().partitioner().numPartitions();
int shuffleId = rssShuffleHandle.getShuffleId();
ShuffleHandleInfo shuffleHandleInfo;
if (rssResubmitStage) {
if (rssResubmitStage || taskFailRetry) {
// Get the ShuffleServer list from the Driver based on the shuffleId
shuffleHandleInfo = getRemoteShuffleHandleInfo(shuffleId);
} else {
Expand Down
3 changes: 2 additions & 1 deletion client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,8 @@ private RssShuffleWriter(
this.sparkConf = sparkConf;
this.taskFailRetry =
sparkConf.getBoolean(
RssClientConf.RSS_TASK_FAILED_RETRY_ENABLED.key(),
RssSparkConfig.SPARK_RSS_CONFIG_PREFIX
+ RssClientConf.RSS_TASK_FAILED_RETRY_ENABLED.key(),
RssClientConf.RSS_TASK_FAILED_RETRY_ENABLED.defaultValue());
}

Expand Down
Loading