You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "zuston (via GitHub)" <gi...@apache.org> on 2024/04/02 06:39:26 UTC
[PR] [#1608][part-3] feat(spark3): support reading from partition block data reassignment servers [incubator-uniffle]
zuston opened a new pull request, #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615
### What changes were proposed in this pull request?
Support reading from partition block data reassignment servers
### Why are the changes needed?
For: #1608
Writer has been writing data into reassignment servers, so it's necessary to read from reassignment servers
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
`PartitionBlockDataReassignTest` integration test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1557017599
##########
proto/src/main/proto/Rss.proto:
##########
@@ -551,9 +551,8 @@ message PartitionToShuffleServerRequest {
message PartitionToShuffleServerResponse {
StatusCode status = 1;
- map<int32,GetShuffleServerListResponse> partitionToShuffleServer = 2;
- RemoteStorageInfo remote_storage_info = 3;
- string msg = 4;
+ string msg = 2;
+ bytes shuffleHandleInfoSerializableBytes = 3;
Review Comment:
Why do we need this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading from multiple reassignment servers [incubator-uniffle]
Posted by "xumanbu (via GitHub)" <gi...@apache.org>.
xumanbu commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1555184982
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -663,22 +664,17 @@ public <K, C> ShuffleReader<K, C> getReaderImpl(
rssShuffleHandle.getPartitionToServers(),
rssShuffleHandle.getRemoteStorage());
}
- Map<Integer, List<ShuffleServerInfo>> allPartitionToServers =
- shuffleHandleInfo.getPartitionToServers();
- Map<Integer, List<ShuffleServerInfo>> requirePartitionToServers =
- allPartitionToServers.entrySet().stream()
- .filter(x -> x.getKey() >= startPartition && x.getKey() < endPartition)
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
- Map<ShuffleServerInfo, Set<Integer>> serverToPartitions =
- RssUtils.generateServerToPartitions(requirePartitionToServers);
+ Map<ShuffleServerInfo, Set<Integer>> blockIdServerToPartitions =
Review Comment:
why was `serverToPartitions` rename to `blockIdServerToPartitions` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2044278465
> There are some points which I cared:
>
> 1. Do we support multiple times reassignments? Do we add some tests for this case?
Multiple reassignments servers for one partition will be added in the next PR, which will be used by the following cases;
1. reassign servers fails to send blocks and then reassign again for one partition
2. partition is marked as a huge partition, which will be reassign to multiple servers at the first time, and then different tasks using different reassignment partition server by hash.
> 2. Do we work with multiple replicas well?
This should be more tests. But in this stage, I will make `multiple replicas` and `reassign` exlusive by the extra check, which also will be added in the next PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "dingshun3016 (via GitHub)" <gi...@apache.org>.
dingshun3016 commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1555617142
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/ShuffleHandleInfo.java:
##########
@@ -97,21 +105,83 @@ public ShuffleServerInfo useExistingReassignmentForMultiPartitions(
public ShuffleServerInfo createNewReassignmentForMultiPartitions(
Set<Integer> partitionIds, String faultyServerId, ShuffleServerInfo replacement) {
if (replacement != null) {
- faultyServerReplacements.put(faultyServerId, replacement);
+ faultyServerToReplacements.put(faultyServerId, replacement);
}
- replacement = faultyServerReplacements.get(faultyServerId);
+ replacement = faultyServerToReplacements.get(faultyServerId);
for (Integer partitionId : partitionIds) {
List<ShuffleServerInfo> replicaServers = partitionToServers.get(partitionId);
for (int i = 0; i < replicaServers.size(); i++) {
if (replicaServers.get(i).getId().equals(faultyServerId)) {
Review Comment:
There a little problem, when reassign server for the second time, faultyServerId is not in partitionToServers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading from multiple reassignment servers [incubator-uniffle]
Posted by "xumanbu (via GitHub)" <gi...@apache.org>.
xumanbu commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1555184982
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -663,22 +664,17 @@ public <K, C> ShuffleReader<K, C> getReaderImpl(
rssShuffleHandle.getPartitionToServers(),
rssShuffleHandle.getRemoteStorage());
}
- Map<Integer, List<ShuffleServerInfo>> allPartitionToServers =
- shuffleHandleInfo.getPartitionToServers();
- Map<Integer, List<ShuffleServerInfo>> requirePartitionToServers =
- allPartitionToServers.entrySet().stream()
- .filter(x -> x.getKey() >= startPartition && x.getKey() < endPartition)
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
- Map<ShuffleServerInfo, Set<Integer>> serverToPartitions =
- RssUtils.generateServerToPartitions(requirePartitionToServers);
+ Map<ShuffleServerInfo, Set<Integer>> blockIdServerToPartitions =
Review Comment:
`serverToPartitions` is a better name than `blockIdServerToPartitions`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading from partition block data reassignment servers [incubator-uniffle]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2031201963
## [Codecov](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1615?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `88.88889%` with `2 lines` in your changes are missing coverage. Please review.
> Project coverage is 54.96%. Comparing base [(`1051d26`)](https://app.codecov.io/gh/apache/incubator-uniffle/commit/1051d26352b99a533e808b78835e26b338852331?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`7dfdb43`)](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1615?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
| [Files](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1615?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...he/uniffle/server/buffer/ShuffleBufferManager.java](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1615?src=pr&el=tree&filepath=server%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Funiffle%2Fserver%2Fbuffer%2FShuffleBufferManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9idWZmZXIvU2h1ZmZsZUJ1ZmZlck1hbmFnZXIuamF2YQ==) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1615?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #1615 +/- ##
============================================
+ Coverage 53.98% 54.96% +0.98%
- Complexity 2872 2881 +9
============================================
Files 438 418 -20
Lines 24927 22584 -2343
Branches 2126 2130 +4
============================================
- Hits 13456 12414 -1042
+ Misses 10626 9398 -1228
+ Partials 845 772 -73
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1615?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1568117273
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/ShuffleHandleInfo.java:
##########
@@ -85,33 +126,125 @@ public int getShuffleId() {
return shuffleId;
}
- public boolean isExistingFaultyServer(String serverId) {
- return faultyServerReplacements.containsKey(serverId);
+ @VisibleForTesting
+ protected boolean isMarkedAsFaultyServer(String serverId) {
+ return faultyServerToReplacements.containsKey(serverId);
}
- public ShuffleServerInfo useExistingReassignmentForMultiPartitions(
- Set<Integer> partitionIds, String faultyServerId) {
- return createNewReassignmentForMultiPartitions(partitionIds, faultyServerId, null);
+ public Set<ShuffleServerInfo> getExistingReplacements(String faultyServerId) {
+ return faultyServerToReplacements.get(faultyServerId);
}
- public ShuffleServerInfo createNewReassignmentForMultiPartitions(
- Set<Integer> partitionIds, String faultyServerId, ShuffleServerInfo replacement) {
- if (replacement != null) {
- faultyServerReplacements.put(faultyServerId, replacement);
+ public void updateReassignment(
+ Set<Integer> partitionIds, String faultyServerId, Set<ShuffleServerInfo> replacements) {
+ if (replacements == null) {
+ return;
}
- replacement = faultyServerReplacements.get(faultyServerId);
+ faultyServerToReplacements.put(faultyServerId, replacements);
+ // todo: optimize the multiple for performance
for (Integer partitionId : partitionIds) {
- List<ShuffleServerInfo> replicaServers = partitionToServers.get(partitionId);
- for (int i = 0; i < replicaServers.size(); i++) {
- if (replicaServers.get(i).getId().equals(faultyServerId)) {
- Map<Integer, ShuffleServerInfo> replicaReplacements =
- failoverPartitionServers.computeIfAbsent(
- partitionId, k -> JavaUtils.newConcurrentMap());
- replicaReplacements.put(i, replacement);
+ Map<Integer, List<ShuffleServerInfo>> replicaServers =
+ partitionReplicaAssignedServers.get(partitionId);
+ for (Map.Entry<Integer, List<ShuffleServerInfo>> serverEntry : replicaServers.entrySet()) {
+ List<ShuffleServerInfo> servers = serverEntry.getValue();
+ if (servers.stream()
+ .map(x -> x.getId())
+ .collect(Collectors.toSet())
+ .contains(faultyServerId)) {
+ Set<ShuffleServerInfo> tempSet = new HashSet<>();
+ tempSet.addAll(replacements);
+ tempSet.removeAll(servers);
+ servers.addAll(tempSet);
Review Comment:
Yes.
The name of `faulty servers` here is not accurate, (But here I will not change this, maybe later.) because due to tight memory, the NO_BUFFER will be threw. For these cases, we'd better to reassign partition server to write data.
So the reserving these server is to fetch partial data when reading.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1558685271
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/KryoSerializerWrapper.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.shuffle;
+
+import java.nio.ByteBuffer;
+
+import scala.reflect.ClassTag$;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.serializer.KryoSerializer;
+import org.apache.spark.serializer.SerializerInstance;
+
+public class KryoSerializerWrapper {
Review Comment:
Yes similar implementations could be acted in shuffleManager rpc service, because the driver and executor are always bounded into the same version, no compatiblity problems.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2042290991
PTAL @dingshun3016 @xumanbu @jerqi
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1556939969
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/ShuffleHandleInfo.java:
##########
@@ -97,21 +105,83 @@ public ShuffleServerInfo useExistingReassignmentForMultiPartitions(
public ShuffleServerInfo createNewReassignmentForMultiPartitions(
Set<Integer> partitionIds, String faultyServerId, ShuffleServerInfo replacement) {
if (replacement != null) {
- faultyServerReplacements.put(faultyServerId, replacement);
+ faultyServerToReplacements.put(faultyServerId, replacement);
}
- replacement = faultyServerReplacements.get(faultyServerId);
+ replacement = faultyServerToReplacements.get(faultyServerId);
for (Integer partitionId : partitionIds) {
List<ShuffleServerInfo> replicaServers = partitionToServers.get(partitionId);
for (int i = 0; i < replicaServers.size(); i++) {
if (replicaServers.get(i).getId().equals(faultyServerId)) {
Review Comment:
Currently, the server reassignment will only occur once time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "dingshun3016 (via GitHub)" <gi...@apache.org>.
dingshun3016 commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2051064760
In the same task, when the failed blocks resend successfully and the remaining blocks need to be sent, `partitionToServers` needs to be replaced with the last one in `org.apache.spark.shuffle.writer.WriteBufferManager#createShuffleBlock`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1557017316
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/KryoSerializerWrapper.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.shuffle;
+
+import java.nio.ByteBuffer;
+
+import scala.reflect.ClassTag$;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.serializer.KryoSerializer;
+import org.apache.spark.serializer.SerializerInstance;
+
+public class KryoSerializerWrapper {
Review Comment:
We shouldn't bind to KryoSeriliazerWrapper.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1567457776
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/ShuffleHandleInfo.java:
##########
@@ -85,33 +126,125 @@ public int getShuffleId() {
return shuffleId;
}
- public boolean isExistingFaultyServer(String serverId) {
- return faultyServerReplacements.containsKey(serverId);
+ @VisibleForTesting
+ protected boolean isMarkedAsFaultyServer(String serverId) {
+ return faultyServerToReplacements.containsKey(serverId);
}
- public ShuffleServerInfo useExistingReassignmentForMultiPartitions(
- Set<Integer> partitionIds, String faultyServerId) {
- return createNewReassignmentForMultiPartitions(partitionIds, faultyServerId, null);
+ public Set<ShuffleServerInfo> getExistingReplacements(String faultyServerId) {
+ return faultyServerToReplacements.get(faultyServerId);
}
- public ShuffleServerInfo createNewReassignmentForMultiPartitions(
- Set<Integer> partitionIds, String faultyServerId, ShuffleServerInfo replacement) {
- if (replacement != null) {
- faultyServerReplacements.put(faultyServerId, replacement);
+ public void updateReassignment(
+ Set<Integer> partitionIds, String faultyServerId, Set<ShuffleServerInfo> replacements) {
+ if (replacements == null) {
+ return;
}
- replacement = faultyServerReplacements.get(faultyServerId);
+ faultyServerToReplacements.put(faultyServerId, replacements);
+ // todo: optimize the multiple for performance
for (Integer partitionId : partitionIds) {
- List<ShuffleServerInfo> replicaServers = partitionToServers.get(partitionId);
- for (int i = 0; i < replicaServers.size(); i++) {
- if (replicaServers.get(i).getId().equals(faultyServerId)) {
- Map<Integer, ShuffleServerInfo> replicaReplacements =
- failoverPartitionServers.computeIfAbsent(
- partitionId, k -> JavaUtils.newConcurrentMap());
- replicaReplacements.put(i, replacement);
+ Map<Integer, List<ShuffleServerInfo>> replicaServers =
+ partitionReplicaAssignedServers.get(partitionId);
+ for (Map.Entry<Integer, List<ShuffleServerInfo>> serverEntry : replicaServers.entrySet()) {
+ List<ShuffleServerInfo> servers = serverEntry.getValue();
+ if (servers.stream()
+ .map(x -> x.getId())
+ .collect(Collectors.toSet())
+ .contains(faultyServerId)) {
+ Set<ShuffleServerInfo> tempSet = new HashSet<>();
+ tempSet.addAll(replacements);
+ tempSet.removeAll(servers);
+ servers.addAll(tempSet);
Review Comment:
This will keep the faulty servers in `servers`, is this expected?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2058688151
Ping @jerqi
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading from partition block data reassignment servers [incubator-uniffle]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2031221705
## Test Results
2 356 files + 7 2 356 suites +7 4h 32m 58s :stopwatch: + 2m 17s
911 tests + 2 909 :white_check_mark: + 2 1 :zzz: ±0 0 :x: - 1 1 :fire: +1
10 566 runs +16 10 545 :white_check_mark: +10 14 :zzz: ±0 0 :x: - 1 7 :fire: +7
For more details on these errors, see [this check](https://github.com/apache/incubator-uniffle/runs/23331425806).
Results for commit 7dfdb431. ± Comparison against base commit 1051d263.
[test-results]:data:application/gzip;base64,H4sIAI6tC2YC/03MTQ7CIBCG4asY1i7AGYbWyxgKJSH2x1BYGe/uYErt8nsY3rcIcRo3cb/cQNP1IrYS83n7kmyO68KiCEzHxO+5XvRKtfXYinOVZP+nZ3zVXwcEGycGecCY0pr2k1SW2lRSE+2zRdlQH7ZXscGp+tstahjcOs8x8xDGBz8gKNMjBiDpcRg7kmQdIPhOBm0HMF6LzxfA1/TvEgEAAA==
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "xumanbu (via GitHub)" <gi...@apache.org>.
xumanbu commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1558805992
##########
server/src/main/java/org/apache/uniffle/server/buffer/ShuffleBufferManager.java:
##########
@@ -712,4 +712,8 @@ public boolean limitHugePartition(
}
return false;
}
+
+ public void setUsedMemory(long usedMemory) {
Review Comment:
This function should only be used for testing purposes ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2054336841
Fix the CI failure. PTAL @jerqi
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1568774511
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/ShuffleHandleInfo.java:
##########
@@ -85,33 +126,125 @@ public int getShuffleId() {
return shuffleId;
}
- public boolean isExistingFaultyServer(String serverId) {
- return faultyServerReplacements.containsKey(serverId);
+ @VisibleForTesting
+ protected boolean isMarkedAsFaultyServer(String serverId) {
+ return faultyServerToReplacements.containsKey(serverId);
}
- public ShuffleServerInfo useExistingReassignmentForMultiPartitions(
- Set<Integer> partitionIds, String faultyServerId) {
- return createNewReassignmentForMultiPartitions(partitionIds, faultyServerId, null);
+ public Set<ShuffleServerInfo> getExistingReplacements(String faultyServerId) {
+ return faultyServerToReplacements.get(faultyServerId);
}
- public ShuffleServerInfo createNewReassignmentForMultiPartitions(
- Set<Integer> partitionIds, String faultyServerId, ShuffleServerInfo replacement) {
- if (replacement != null) {
- faultyServerReplacements.put(faultyServerId, replacement);
+ public void updateReassignment(
+ Set<Integer> partitionIds, String faultyServerId, Set<ShuffleServerInfo> replacements) {
+ if (replacements == null) {
+ return;
}
- replacement = faultyServerReplacements.get(faultyServerId);
+ faultyServerToReplacements.put(faultyServerId, replacements);
+ // todo: optimize the multiple for performance
for (Integer partitionId : partitionIds) {
- List<ShuffleServerInfo> replicaServers = partitionToServers.get(partitionId);
- for (int i = 0; i < replicaServers.size(); i++) {
- if (replicaServers.get(i).getId().equals(faultyServerId)) {
- Map<Integer, ShuffleServerInfo> replicaReplacements =
- failoverPartitionServers.computeIfAbsent(
- partitionId, k -> JavaUtils.newConcurrentMap());
- replicaReplacements.put(i, replacement);
+ Map<Integer, List<ShuffleServerInfo>> replicaServers =
+ partitionReplicaAssignedServers.get(partitionId);
+ for (Map.Entry<Integer, List<ShuffleServerInfo>> serverEntry : replicaServers.entrySet()) {
+ List<ShuffleServerInfo> servers = serverEntry.getValue();
+ if (servers.stream()
+ .map(x -> x.getId())
+ .collect(Collectors.toSet())
+ .contains(faultyServerId)) {
+ Set<ShuffleServerInfo> tempSet = new HashSet<>();
+ tempSet.addAll(replacements);
+ tempSet.removeAll(servers);
+ servers.addAll(tempSet);
Review Comment:
Oh, I think I need to say the reassign occurs on the writer side. Sorry I still don’t catch your thought.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1560768012
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/KryoSerializerWrapper.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.shuffle;
+
+import java.nio.ByteBuffer;
+
+import scala.reflect.ClassTag$;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.serializer.KryoSerializer;
+import org.apache.spark.serializer.SerializerInstance;
+
+public class KryoSerializerWrapper {
Review Comment:
This has been refactored to avoid de/serialization performance drop
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/KryoSerializerWrapper.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.shuffle;
+
+import java.nio.ByteBuffer;
+
+import scala.reflect.ClassTag$;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.serializer.KryoSerializer;
+import org.apache.spark.serializer.SerializerInstance;
+
+public class KryoSerializerWrapper {
Review Comment:
This has been refactored to avoid de/serialization performance drop @jerqi
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston merged PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2044263453
There are some points which I cared:
1. Do we support multiple times reassignments? Do we add some tests for this case?
2. Do we work with multiple replicas well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1557105553
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/KryoSerializerWrapper.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.shuffle;
+
+import java.nio.ByteBuffer;
+
+import scala.reflect.ClassTag$;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.serializer.KryoSerializer;
+import org.apache.spark.serializer.SerializerInstance;
+
+public class KryoSerializerWrapper {
Review Comment:
I hope the shuffleHandleInfo could be shared by the driver and executors by seriliazed bytes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "xumanbu (via GitHub)" <gi...@apache.org>.
xumanbu commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1557499150
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/KryoSerializerWrapper.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.shuffle;
+
+import java.nio.ByteBuffer;
+
+import scala.reflect.ClassTag$;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.serializer.KryoSerializer;
+import org.apache.spark.serializer.SerializerInstance;
+
+public class KryoSerializerWrapper {
Review Comment:
Can other RPCs also be implemented in this manner?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1558892739
##########
server/src/main/java/org/apache/uniffle/server/buffer/ShuffleBufferManager.java:
##########
@@ -712,4 +712,8 @@ public boolean limitHugePartition(
}
return false;
}
+
+ public void setUsedMemory(long usedMemory) {
Review Comment:
Yes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1568321670
##########
client/src/main/java/org/apache/uniffle/client/PartitionDataReplicaRequirementTracking.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.
+ */
+
+package org.apache.uniffle.client;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.uniffle.common.ShuffleServerInfo;
+
+/**
+ * This class is to track the partition data replica requirements, which is used for {@link
+ * org.apache.uniffle.client.impl.ShuffleWriteClientImpl} to check whether the reading blockIds from
+ * multi/single server(s) meet the min replica.
+ */
+public class PartitionDataReplicaRequirementTracking {
+ private int shuffleId;
+
+ // partitionId -> replicaIndex -> shuffleServerInfo
+ private Map<Integer, Map<Integer, List<ShuffleServerInfo>>> inventory;
+
+ private Map<Integer, Map<Integer, Integer>> succeedList = new HashMap<>();
+
+ public PartitionDataReplicaRequirementTracking(
+ int shuffleId, Map<Integer, Map<Integer, List<ShuffleServerInfo>>> inventory) {
+ this.shuffleId = shuffleId;
+ this.inventory = inventory;
+ }
+
+ public boolean isSatisfied(int partitionId, int minReplica) {
+ // replica index -> successful count
+ Map<Integer, Integer> succeedReplicas = succeedList.get(partitionId);
+ if (succeedReplicas == null) {
+ succeedReplicas = new HashMap<>();
+ }
+
+ Map<Integer, List<ShuffleServerInfo>> replicaList = inventory.get(partitionId);
Review Comment:
What if `partitionId` does not exist?
##########
client/src/main/java/org/apache/uniffle/client/PartitionDataReplicaRequirementTracking.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.
+ */
+
+package org.apache.uniffle.client;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.uniffle.common.ShuffleServerInfo;
+
+/**
+ * This class is to track the partition data replica requirements, which is used for {@link
+ * org.apache.uniffle.client.impl.ShuffleWriteClientImpl} to check whether the reading blockIds from
+ * multi/single server(s) meet the min replica.
+ */
+public class PartitionDataReplicaRequirementTracking {
+ private int shuffleId;
+
+ // partitionId -> replicaIndex -> shuffleServerInfo
+ private Map<Integer, Map<Integer, List<ShuffleServerInfo>>> inventory;
+
+ private Map<Integer, Map<Integer, Integer>> succeedList = new HashMap<>();
+
+ public PartitionDataReplicaRequirementTracking(
+ int shuffleId, Map<Integer, Map<Integer, List<ShuffleServerInfo>>> inventory) {
+ this.shuffleId = shuffleId;
+ this.inventory = inventory;
+ }
+
+ public boolean isSatisfied(int partitionId, int minReplica) {
+ // replica index -> successful count
+ Map<Integer, Integer> succeedReplicas = succeedList.get(partitionId);
+ if (succeedReplicas == null) {
+ succeedReplicas = new HashMap<>();
+ }
Review Comment:
This can be simplified:
```
Map<Integer, Integer> succeedReplicas = succeedList.getOrDefault(partitionId, new HashMap<>());
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1568316424
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/ShuffleHandleInfo.java:
##########
@@ -85,33 +126,125 @@ public int getShuffleId() {
return shuffleId;
}
- public boolean isExistingFaultyServer(String serverId) {
- return faultyServerReplacements.containsKey(serverId);
+ @VisibleForTesting
+ protected boolean isMarkedAsFaultyServer(String serverId) {
+ return faultyServerToReplacements.containsKey(serverId);
}
- public ShuffleServerInfo useExistingReassignmentForMultiPartitions(
- Set<Integer> partitionIds, String faultyServerId) {
- return createNewReassignmentForMultiPartitions(partitionIds, faultyServerId, null);
+ public Set<ShuffleServerInfo> getExistingReplacements(String faultyServerId) {
+ return faultyServerToReplacements.get(faultyServerId);
}
- public ShuffleServerInfo createNewReassignmentForMultiPartitions(
- Set<Integer> partitionIds, String faultyServerId, ShuffleServerInfo replacement) {
- if (replacement != null) {
- faultyServerReplacements.put(faultyServerId, replacement);
+ public void updateReassignment(
+ Set<Integer> partitionIds, String faultyServerId, Set<ShuffleServerInfo> replacements) {
+ if (replacements == null) {
+ return;
}
- replacement = faultyServerReplacements.get(faultyServerId);
+ faultyServerToReplacements.put(faultyServerId, replacements);
+ // todo: optimize the multiple for performance
for (Integer partitionId : partitionIds) {
- List<ShuffleServerInfo> replicaServers = partitionToServers.get(partitionId);
- for (int i = 0; i < replicaServers.size(); i++) {
- if (replicaServers.get(i).getId().equals(faultyServerId)) {
- Map<Integer, ShuffleServerInfo> replicaReplacements =
- failoverPartitionServers.computeIfAbsent(
- partitionId, k -> JavaUtils.newConcurrentMap());
- replicaReplacements.put(i, replacement);
+ Map<Integer, List<ShuffleServerInfo>> replicaServers =
+ partitionReplicaAssignedServers.get(partitionId);
+ for (Map.Entry<Integer, List<ShuffleServerInfo>> serverEntry : replicaServers.entrySet()) {
+ List<ShuffleServerInfo> servers = serverEntry.getValue();
+ if (servers.stream()
+ .map(x -> x.getId())
+ .collect(Collectors.toSet())
+ .contains(faultyServerId)) {
+ Set<ShuffleServerInfo> tempSet = new HashSet<>();
+ tempSet.addAll(replacements);
+ tempSet.removeAll(servers);
+ servers.addAll(tempSet);
Review Comment:
Why don't we read from all servers until all blocks are read? Then we would not need to add replacement server on failure here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1555258945
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -663,22 +664,17 @@ public <K, C> ShuffleReader<K, C> getReaderImpl(
rssShuffleHandle.getPartitionToServers(),
rssShuffleHandle.getRemoteStorage());
}
- Map<Integer, List<ShuffleServerInfo>> allPartitionToServers =
- shuffleHandleInfo.getPartitionToServers();
- Map<Integer, List<ShuffleServerInfo>> requirePartitionToServers =
- allPartitionToServers.entrySet().stream()
- .filter(x -> x.getKey() >= startPartition && x.getKey() < endPartition)
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
- Map<ShuffleServerInfo, Set<Integer>> serverToPartitions =
- RssUtils.generateServerToPartitions(requirePartitionToServers);
+ Map<ShuffleServerInfo, Set<Integer>> blockIdServerToPartitions =
Review Comment:
My fault. I will recover this part after #1610 is merged/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#discussion_r1555518391
##########
integration-test/spark3/src/test/java/org/apache/uniffle/test/PartitionBlockDataReassignTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.
+ */
+
+package org.apache.uniffle.test;
+
+import java.io.File;
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+import org.apache.spark.SparkConf;
+import org.apache.spark.shuffle.RssSparkConfig;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.uniffle.common.rpc.ServerType;
+import org.apache.uniffle.coordinator.CoordinatorConf;
+import org.apache.uniffle.server.ShuffleServer;
+import org.apache.uniffle.server.ShuffleServerConf;
+import org.apache.uniffle.server.buffer.ShuffleBufferManager;
+import org.apache.uniffle.storage.util.StorageType;
+
+import static org.apache.uniffle.client.util.RssClientConfig.RSS_CLIENT_ASSIGNMENT_SHUFFLE_SERVER_NUMBER;
+import static org.apache.uniffle.common.config.RssClientConf.RSS_CLIENT_BLOCK_SEND_FAILURE_RETRY_ENABLED;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** This class is to test the mechanism of partition block data reassignment. */
+public class PartitionBlockDataReassignTest extends SparkSQLTest {
Review Comment:
This class is to test the reassign spark test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2051669939
> In the same task, when the failed blocks resend successfully and the remaining blocks need to be sent, `partitionToServers` needs to be replaced with the last one in `org.apache.spark.shuffle.writer.WriteBufferManager#createShuffleBlock`
Yes, this is on the plan, but i will not reuse the concept of partitionToServer. Please wait for the following PRs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1608][part-3] feat(spark3): support reading partition data from multiple reassignment servers [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1615:
URL: https://github.com/apache/incubator-uniffle/pull/1615#issuecomment-2060207135
Thanks @jerqi @dingshun3016 @xumanbu @EnricoMi for your review. Merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org