You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/11/08 12:12:28 UTC
[GitHub] [spark] yabola opened a new pull request, #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
yabola opened a new pull request, #38560:
URL: https://github.com/apache/spark/pull/38560
I make a PR first for review quickly to avoid some big mistakes. At present, this function is only roughly implemented, I still need to improve the code and supplement UT.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1308437041
I am wondering whether the driver needs to pass the merged reduceId to the external shuffle service, or the shuffle service records the merged reduceIds, and the subsequent driver nodes only need to pass the shuffleId and other information
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1321687314
> @mridulm @wankunde @otterc I'm not sure if I missed any logic, please help review my code , thanks~ I will improve my code style later. Now I don't change my code in `BlockManagerMasterEndpoint` as #37922 do . Can it be split into two PRs, I implement the code of the shuffle service part first and @wankunde finish the rest part since he has done?
Some thoughts on these changes I wrote 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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1363532585
close on https://github.com/apache/spark/pull/37922
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1023816561
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -654,8 +731,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
// If no blocks pushed for the finalizeShuffleMerge shuffleMergeId then return
// empty MergeStatuses but cleanup the older shuffleMergeId files.
submitCleanupTask(() ->
- closeAndDeleteOutdatedPartitions(
- appAttemptShuffleMergeId, mergePartitionsInfo.shuffleMergePartitions));
+ deleteCurrentShufflePartitions(appShuffleInfo, shuffleId, appAttemptShuffleMergeId));
Review Comment:
I checked the `appAttemptShuffleMergeId` in the code before.
I think if we want to delete partitions merged data, then we must delete the corresponding ShuffleMergeId in DB (Otherwise, inconsistency will occur when restoring shuffle info from db)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1023828427
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -654,8 +731,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
// If no blocks pushed for the finalizeShuffleMerge shuffleMergeId then return
// empty MergeStatuses but cleanup the older shuffleMergeId files.
submitCleanupTask(() ->
- closeAndDeleteOutdatedPartitions(
- appAttemptShuffleMergeId, mergePartitionsInfo.shuffleMergePartitions));
+ deleteCurrentShufflePartitions(appShuffleInfo, shuffleId, appAttemptShuffleMergeId));
Review Comment:
And I think when deleting partition information, we may don’t store shuffleMergeID in DB anymore
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1030998613
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -327,43 +327,52 @@ class BlockManagerMasterEndpoint(
}
}.toSeq
- // Find all shuffle blocks on executors that are no longer running
- val blocksToDeleteByShuffleService =
- new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ var removeShuffleFromShuffleServicesFutures = Seq.empty[Future[Boolean]]
if (externalShuffleServiceRemoveShuffleEnabled) {
- mapOutputTracker.shuffleStatuses.get(shuffleId).foreach { shuffleStatus =>
- shuffleStatus.withMapStatuses { mapStatuses =>
- mapStatuses.foreach { mapStatus =>
- // Check if the executor has been deallocated
- if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
- val blocksToDel =
- shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId, mapStatus.mapId)
- if (blocksToDel.nonEmpty) {
- val blocks = blocksToDeleteByShuffleService.getOrElseUpdate(mapStatus.location,
- new mutable.HashSet[BlockId])
- blocks ++= blocksToDel
+ val shuffleClient = externalBlockStoreClient.get
+ // Find all shuffle blocks on executors that are no longer running
+ val blocksToDelete = new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ mapOutputTracker.shuffleStatuses.get(shuffleId) match {
+ case Some(shuffleStatus) =>
+ shuffleStatus.withMapStatuses { mapStatuses =>
+ mapStatuses.foreach { mapStatus =>
+ // Check if the executor has been deallocated
+ if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
+ val blocksToDel = shuffleManager.shuffleBlockResolver
+ .getBlocksForShuffle(shuffleId, mapStatus.mapId)
+ if (blocksToDel.nonEmpty) {
+ val blocks = blocksToDelete.getOrElseUpdate(mapStatus.location,
+ new mutable.HashSet[BlockId])
+ blocks ++= blocksToDel
+ }
Review Comment:
> What if the shuffle statuses are not exists ?
I think it will not, please see [case match codes](https://github.com/apache/spark/pull/38560/files#diff-21bb40987f7d21c8d3d3de3212f388c210fb04a636c96bdef53cab6070c6aa46R336)
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -327,43 +327,52 @@ class BlockManagerMasterEndpoint(
}
}.toSeq
- // Find all shuffle blocks on executors that are no longer running
- val blocksToDeleteByShuffleService =
- new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ var removeShuffleFromShuffleServicesFutures = Seq.empty[Future[Boolean]]
if (externalShuffleServiceRemoveShuffleEnabled) {
- mapOutputTracker.shuffleStatuses.get(shuffleId).foreach { shuffleStatus =>
- shuffleStatus.withMapStatuses { mapStatuses =>
- mapStatuses.foreach { mapStatus =>
- // Check if the executor has been deallocated
- if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
- val blocksToDel =
- shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId, mapStatus.mapId)
- if (blocksToDel.nonEmpty) {
- val blocks = blocksToDeleteByShuffleService.getOrElseUpdate(mapStatus.location,
- new mutable.HashSet[BlockId])
- blocks ++= blocksToDel
+ val shuffleClient = externalBlockStoreClient.get
+ // Find all shuffle blocks on executors that are no longer running
+ val blocksToDelete = new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ mapOutputTracker.shuffleStatuses.get(shuffleId) match {
+ case Some(shuffleStatus) =>
+ shuffleStatus.withMapStatuses { mapStatuses =>
+ mapStatuses.foreach { mapStatus =>
+ // Check if the executor has been deallocated
+ if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
+ val blocksToDel = shuffleManager.shuffleBlockResolver
+ .getBlocksForShuffle(shuffleId, mapStatus.mapId)
+ if (blocksToDel.nonEmpty) {
+ val blocks = blocksToDelete.getOrElseUpdate(mapStatus.location,
+ new mutable.HashSet[BlockId])
+ blocks ++= blocksToDel
+ }
Review Comment:
> What if the shuffle statuses are not exists ?
I think it will not, please see [case match codes](https://github.com/apache/spark/pull/38560/files#diff-21bb40987f7d21c8d3d3de3212f388c210fb04a636c96bdef53cab6070c6aa46R336)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1321726307
Thanks for taking over the PR @yabola !
I am heading out on a vacation soon, not sure if @otterc will have bandwidth to take a look in meantime.
I will definitely circle back to this once you are done, and I am back.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1310162375
> I am wondering whether the driver needs to pass the merged reduceId to the external shuffle service (but now the driver cannot fully record merged info), or the shuffle service records the merged reduceIds, and driver only need to pass the shuffleId and other information
I decided to change the driver to not send reduceId, because only the shuffle service finally understands which shuffle data is stored, no matter how the driver processes or processes the message
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1307631649
+CC @otterc
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1023798104
##########
common/network-common/src/main/java/org/apache/spark/network/util/PushBasedShuffleUtils.java:
##########
@@ -0,0 +1,33 @@
+/*
+ * 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.network.util;
+
+public class PushBasedShuffleUtils {
+
+ /**
+ * Push based shuffle requires a comparable Id to distinguish the shuffle data among multiple application attempts.
+ * This variable is derived from the String typed appAttemptId. If no appAttemptId is set, the default
+ * comparableAppAttemptId is -1.
+ */
+ public static final int DEFAUT_APP_ATTEMPT_ID = -1;
Review Comment:
Unified push based shuffle identification variables here, which will be used in yarn external shuffle service and spark core module.
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -327,43 +327,52 @@ class BlockManagerMasterEndpoint(
}
}.toSeq
- // Find all shuffle blocks on executors that are no longer running
- val blocksToDeleteByShuffleService =
- new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ var removeShuffleFromShuffleServicesFutures = Seq.empty[Future[Boolean]]
if (externalShuffleServiceRemoveShuffleEnabled) {
- mapOutputTracker.shuffleStatuses.get(shuffleId).foreach { shuffleStatus =>
- shuffleStatus.withMapStatuses { mapStatuses =>
- mapStatuses.foreach { mapStatus =>
- // Check if the executor has been deallocated
- if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
- val blocksToDel =
- shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId, mapStatus.mapId)
- if (blocksToDel.nonEmpty) {
- val blocks = blocksToDeleteByShuffleService.getOrElseUpdate(mapStatus.location,
- new mutable.HashSet[BlockId])
- blocks ++= blocksToDel
+ val shuffleClient = externalBlockStoreClient.get
+ // Find all shuffle blocks on executors that are no longer running
+ val blocksToDelete = new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ mapOutputTracker.shuffleStatuses.get(shuffleId) match {
+ case Some(shuffleStatus) =>
+ shuffleStatus.withMapStatuses { mapStatuses =>
+ mapStatuses.foreach { mapStatus =>
+ // Check if the executor has been deallocated
+ if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
+ val blocksToDel = shuffleManager.shuffleBlockResolver
+ .getBlocksForShuffle(shuffleId, mapStatus.mapId)
+ if (blocksToDel.nonEmpty) {
+ val blocks = blocksToDelete.getOrElseUpdate(mapStatus.location,
+ new mutable.HashSet[BlockId])
+ blocks ++= blocksToDel
+ }
Review Comment:
The previous code is difficult to understand because of the scala syntax. For example, `mapOutputTracker.shuffleStatuses.get(shuffleId).**foreach**` The **foreach** here is not actually an iterator.
`externalBlockStoreClient.map` the **externalBlockStoreClient** is not actually an iterator.
I didn't change the code logic, just changed the style
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -654,8 +731,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
// If no blocks pushed for the finalizeShuffleMerge shuffleMergeId then return
// empty MergeStatuses but cleanup the older shuffleMergeId files.
submitCleanupTask(() ->
- closeAndDeleteOutdatedPartitions(
- appAttemptShuffleMergeId, mergePartitionsInfo.shuffleMergePartitions));
+ deleteCurrentShufflePartitions(appShuffleInfo, shuffleId, appAttemptShuffleMergeId));
Review Comment:
I checked the `appAttemptShuffleMergeId` in the code before.
I think if we want to delete partitions merged data, then we must delete the corresponding ShuffleMergeId in DB (Otherwise, inconsistency will occur when restoring shuffle info from db and shufflemergeid is only generated when finalized)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1030393536
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -712,6 +794,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
mergeStatuses = new MergeStatuses(msg.shuffleId, msg.shuffleMergeId,
bitmaps.toArray(new RoaringBitmap[bitmaps.size()]), Ints.toArray(reduceIds),
Longs.toArray(sizes));
+ appShuffleInfo.shuffles.get(msg.shuffleId).setFinalizedPartitions(Ints.toArray(reduceIds));
Review Comment:
Thanks for your review @wankunde !
Yes, if we need to solve this situation completely, we need to store finalized partitions in db.
But on the other hand, when Application removed, all the merged data will be cleaned up finally.
I'm not sure if we can just ignore this case to simplify the logic since it will finally be cleaned up.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1321786141
> One things that I know need to be addressed are: Some merge data infos are not saved on the driver because they are too small ( controlled by `spark.shuffle.push.minShuffleSizeToWait`) please see https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2295
@mridulm sorry, in my previous implementation, I needed to pass the reduceid to the external shuffle service, but I found a problem, the driver cannot record the complete merged reduceId (see my comment for the reason)...
But I had changed my implementation, so it may not be a problem (we can save merged reduceIds in shuffle service, please see .
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1030393536
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -712,6 +794,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
mergeStatuses = new MergeStatuses(msg.shuffleId, msg.shuffleMergeId,
bitmaps.toArray(new RoaringBitmap[bitmaps.size()]), Ints.toArray(reduceIds),
Longs.toArray(sizes));
+ appShuffleInfo.shuffles.get(msg.shuffleId).setFinalizedPartitions(Ints.toArray(reduceIds));
Review Comment:
Thanks for your review @wankunde ! Yes, if we need to solve this situation completely, we need to store finalized partitions in db.
But on the other hand, when Application removed, all the merged data will be cleaned up finally.
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -452,22 +489,69 @@ void removeAppShuffleInfoFromDB(AppShuffleInfo appShuffleInfo) {
}
/**
- * Clean up all the AppShufflePartitionInfo and the finalized shuffle partitions in DB for
- * a specific shuffleMergeId. This is done since there is a higher shuffleMergeId request made
- * for a shuffleId, therefore clean up older shuffleMergeId partitions. The cleanup will be
- * executed the mergedShuffleCleaner thread.
+ * Clean up the outdated finalized or unfinalized shuffle partitions. The cleanup will be executed in the
+ * mergedShuffleCleaner thread. Two cases need to clean up the shuffle: 1. there is a higher shuffleMergeId request
+ * made for a shuffleId, therefore clean up older shuffleMergeId partitions. 2. Application requires to clean up the
+ * expired or unused specific shuffleId partitions
*/
+ @VisibleForTesting
+ void submitRemoveShuffleMergeTask(
+ AppShuffleInfo shuffleInfo, Integer shuffleId,
+ Optional<AppAttemptShuffleMergeId> higherShuffleMergeId) {
+ AppShuffleMergePartitionsInfo mergePartitionsInfo = shuffleInfo.shuffles.get(shuffleId);
+ AppAttemptShuffleMergeId shuffleMergeId = new AppAttemptShuffleMergeId(
+ shuffleInfo.appId, shuffleInfo.attemptId, shuffleId, mergePartitionsInfo.shuffleMergeId);
+ if (!mergePartitionsInfo.isFinalized()) {
+ Map<Integer, AppShufflePartitionInfo> partitionsToClean = mergePartitionsInfo.shuffleMergePartitions;
+ submitCleanupTask(() ->
+ closeAndDeleteOutdatedPartitions(shuffleMergeId, partitionsToClean, higherShuffleMergeId));
+ } else {
+ int[] partitionsToClean = mergePartitionsInfo.finalizedPartitions;
+ submitCleanupTask(() ->
+ deleteOutdatedFinalizedPartitions(shuffleInfo, shuffleMergeId, partitionsToClean, higherShuffleMergeId));
+ }
+ }
+
@VisibleForTesting
void closeAndDeleteOutdatedPartitions(
- AppAttemptShuffleMergeId appAttemptShuffleMergeId,
- Map<Integer, AppShufflePartitionInfo> partitions) {
- removeAppShufflePartitionInfoFromDB(appAttemptShuffleMergeId);
+ AppAttemptShuffleMergeId shuffleMergeId,
+ Map<Integer, AppShufflePartitionInfo> partitions,
+ Optional<AppAttemptShuffleMergeId> higherShuffleMergeId) {
+ removeAppShufflePartitionInfoFromDB(shuffleMergeId);
+ higherShuffleMergeId.ifPresent(this::removeAppShufflePartitionInfoFromDB);
partitions
- .forEach((partitionId, partitionInfo) -> {
- synchronized (partitionInfo) {
- partitionInfo.closeAllFilesAndDeleteIfNeeded(true);
- }
- });
+ .forEach((partitionId, partitionInfo) -> {
+ synchronized (partitionInfo) {
+ partitionInfo.closeAllFilesAndDeleteIfNeeded(true);
+ }
+ });
+ }
+
+ @VisibleForTesting
+ void deleteOutdatedFinalizedPartitions(
+ AppShuffleInfo shuffleInfo,
+ AppAttemptShuffleMergeId shuffleMergeId,
+ int[] outdatedFinalizedPartitions,
+ Optional<AppAttemptShuffleMergeId> higherShuffleMergeId) {
+ int shuffleId = shuffleMergeId.shuffleId;
+ int mergeId = shuffleMergeId.shuffleMergeId;
+ removeAppShufflePartitionInfoFromDB(shuffleMergeId);
+ higherShuffleMergeId.ifPresent(this::removeAppShufflePartitionInfoFromDB);
+ Arrays.stream(outdatedFinalizedPartitions).forEach(partition -> {
+ try {
+ File metaFile =
+ shuffleInfo.getMergedShuffleMetaFile(shuffleId, mergeId, partition);
+ File indexFile = new File(
+ shuffleInfo.getMergedShuffleIndexFilePath(shuffleId, mergeId, partition));
+ File dataFile =
+ shuffleInfo.getMergedShuffleDataFile(shuffleId, mergeId, partition);
+ metaFile.delete();
+ indexFile.delete();
+ dataFile.delete();
+ } catch (Exception e) {
+ logger.error("Error delete shuffle files for {}", shuffleMergeId, e);
+ }
Review Comment:
done
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1316389795
@mridulm as your comment said https://github.com/apache/spark/pull/37922#discussion_r990763769 , I want to Improve this part of the deletion logic
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1310781064
Can one of the admins verify this patch?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1023816561
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -654,8 +731,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
// If no blocks pushed for the finalizeShuffleMerge shuffleMergeId then return
// empty MergeStatuses but cleanup the older shuffleMergeId files.
submitCleanupTask(() ->
- closeAndDeleteOutdatedPartitions(
- appAttemptShuffleMergeId, mergePartitionsInfo.shuffleMergePartitions));
+ deleteCurrentShufflePartitions(appShuffleInfo, shuffleId, appAttemptShuffleMergeId));
Review Comment:
I checked the `appAttemptShuffleMergeId` in the code before.
I think if we want to delete partitions merged data, then we should delete the corresponding ShuffleMergeId in DB (Otherwise, inconsistency will occur when restoring shuffle info from db)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1030393536
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -712,6 +794,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
mergeStatuses = new MergeStatuses(msg.shuffleId, msg.shuffleMergeId,
bitmaps.toArray(new RoaringBitmap[bitmaps.size()]), Ints.toArray(reduceIds),
Longs.toArray(sizes));
+ appShuffleInfo.shuffles.get(msg.shuffleId).setFinalizedPartitions(Ints.toArray(reduceIds));
Review Comment:
Yes, if we need to solve this situation completely, we need to store finalized partitions in db.
But on the other hand, when Application removed, all the merged data will be cleaned up finally.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1321729362
To add, @wankunde's PR is very close to being done.
One approach would be:
a) We get the protocol changes and the immediate impl from that PR merged (once pending comments are addressed).
b) We can follow it up with other changes to make it more robust. For example, [this issue](https://github.com/apache/spark/pull/38560#issuecomment-1321548579) you identified. There might be others as 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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1311176907
@mridulm Yes...These two issues are the similar. @wankunde Can I continue editing my PR in this Issue?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1321685369
@mridulm
I will speed up to finish the unfinished parts of the previous PR together in this PR.
From your comments in the previous PR https://github.com/apache/spark/pull/37922#discussion_r990753031 and https://github.com/apache/spark/pull/37922#discussion_r990763769 .
I had addressed these in my PR , please help review if it is suitable: https://github.com/apache/spark/pull/38560/files#diff-d544fbb952b61283b3d18ca10a5027528efc4f2f65047130da015ae7754c117fR421
and
https://github.com/apache/spark/pull/38560/files#diff-d544fbb952b61283b3d18ca10a5027528efc4f2f65047130da015ae7754c117fR514
Yes , there are nontrivial overlap between https://github.com/apache/spark/pull/37922 , I can cherry pick some codes and fix the comments in this 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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1307984001
The two things that I know need to be addressed are:
1. Some merge data blocks are not saved on the driver because they are too small ( controlled by `spark.shuffle.push.minShuffleSizeToWait`)
please see https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2295
2. The meaning of the `externalShuffleServiceRemoveShuffleEnabled` parameter is not consistent with my current function, but the name is very similar.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1030397010
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -452,22 +489,69 @@ void removeAppShuffleInfoFromDB(AppShuffleInfo appShuffleInfo) {
}
/**
- * Clean up all the AppShufflePartitionInfo and the finalized shuffle partitions in DB for
- * a specific shuffleMergeId. This is done since there is a higher shuffleMergeId request made
- * for a shuffleId, therefore clean up older shuffleMergeId partitions. The cleanup will be
- * executed the mergedShuffleCleaner thread.
+ * Clean up the outdated finalized or unfinalized shuffle partitions. The cleanup will be executed in the
+ * mergedShuffleCleaner thread. Two cases need to clean up the shuffle: 1. there is a higher shuffleMergeId request
+ * made for a shuffleId, therefore clean up older shuffleMergeId partitions. 2. Application requires to clean up the
+ * expired or unused specific shuffleId partitions
*/
+ @VisibleForTesting
+ void submitRemoveShuffleMergeTask(
+ AppShuffleInfo shuffleInfo, Integer shuffleId,
+ Optional<AppAttemptShuffleMergeId> higherShuffleMergeId) {
+ AppShuffleMergePartitionsInfo mergePartitionsInfo = shuffleInfo.shuffles.get(shuffleId);
+ AppAttemptShuffleMergeId shuffleMergeId = new AppAttemptShuffleMergeId(
+ shuffleInfo.appId, shuffleInfo.attemptId, shuffleId, mergePartitionsInfo.shuffleMergeId);
+ if (!mergePartitionsInfo.isFinalized()) {
+ Map<Integer, AppShufflePartitionInfo> partitionsToClean = mergePartitionsInfo.shuffleMergePartitions;
+ submitCleanupTask(() ->
+ closeAndDeleteOutdatedPartitions(shuffleMergeId, partitionsToClean, higherShuffleMergeId));
+ } else {
+ int[] partitionsToClean = mergePartitionsInfo.finalizedPartitions;
+ submitCleanupTask(() ->
+ deleteOutdatedFinalizedPartitions(shuffleInfo, shuffleMergeId, partitionsToClean, higherShuffleMergeId));
+ }
+ }
+
@VisibleForTesting
void closeAndDeleteOutdatedPartitions(
- AppAttemptShuffleMergeId appAttemptShuffleMergeId,
- Map<Integer, AppShufflePartitionInfo> partitions) {
- removeAppShufflePartitionInfoFromDB(appAttemptShuffleMergeId);
+ AppAttemptShuffleMergeId shuffleMergeId,
+ Map<Integer, AppShufflePartitionInfo> partitions,
+ Optional<AppAttemptShuffleMergeId> higherShuffleMergeId) {
+ removeAppShufflePartitionInfoFromDB(shuffleMergeId);
+ higherShuffleMergeId.ifPresent(this::removeAppShufflePartitionInfoFromDB);
partitions
- .forEach((partitionId, partitionInfo) -> {
- synchronized (partitionInfo) {
- partitionInfo.closeAllFilesAndDeleteIfNeeded(true);
- }
- });
+ .forEach((partitionId, partitionInfo) -> {
+ synchronized (partitionInfo) {
+ partitionInfo.closeAllFilesAndDeleteIfNeeded(true);
+ }
+ });
+ }
+
+ @VisibleForTesting
+ void deleteOutdatedFinalizedPartitions(
+ AppShuffleInfo shuffleInfo,
+ AppAttemptShuffleMergeId shuffleMergeId,
+ int[] outdatedFinalizedPartitions,
+ Optional<AppAttemptShuffleMergeId> higherShuffleMergeId) {
+ int shuffleId = shuffleMergeId.shuffleId;
+ int mergeId = shuffleMergeId.shuffleMergeId;
+ removeAppShufflePartitionInfoFromDB(shuffleMergeId);
+ higherShuffleMergeId.ifPresent(this::removeAppShufflePartitionInfoFromDB);
+ Arrays.stream(outdatedFinalizedPartitions).forEach(partition -> {
+ try {
+ File metaFile =
+ shuffleInfo.getMergedShuffleMetaFile(shuffleId, mergeId, partition);
+ File indexFile = new File(
+ shuffleInfo.getMergedShuffleIndexFilePath(shuffleId, mergeId, partition));
+ File dataFile =
+ shuffleInfo.getMergedShuffleDataFile(shuffleId, mergeId, partition);
+ metaFile.delete();
+ indexFile.delete();
+ dataFile.delete();
+ } catch (Exception e) {
+ logger.error("Error delete shuffle files for {}", shuffleMergeId, e);
+ }
Review Comment:
Thanks for your review~ done
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1031346915
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -319,51 +319,68 @@ class BlockManagerMasterEndpoint(
}
private def removeShuffle(shuffleId: Int): Future[Seq[Boolean]] = {
- val removeMsg = RemoveShuffle(shuffleId)
- val removeShuffleFromExecutorsFutures = blockManagerInfo.values.map { bm =>
- bm.storageEndpoint.ask[Boolean](removeMsg).recover {
- // use false as default value means no shuffle data were removed
- handleBlockRemovalFailure("shuffle", shuffleId.toString, bm.blockManagerId, false)
+ var removeShuffleFromShuffleServicesFutures = Seq.empty[Future[Boolean]]
+ if (pushBasedShuffleEnabled && externalBlockStoreClient.isDefined) {
+ val shuffleClient = externalBlockStoreClient.get
+ mapOutputTracker.shuffleStatuses.get(shuffleId) match {
+ case Some(shuffleStatus) =>
+ val shuffleMergerLocations = shuffleStatus.getShufflePushMergerLocations
+ removeShuffleFromShuffleServicesFutures ++= shuffleMergerLocations.map(bmId => {
+ Future[Boolean] {
+ shuffleClient.removeShuffleMerge(bmId.host, bmId.port, shuffleId)
+ }
+ })
+ case None =>
+ logWarning(s"Asked to remove merge shuffle blocks from " +
+ s"shuffle service for unknown shuffle ${shuffleId}")
}
- }.toSeq
+ }
- // Find all shuffle blocks on executors that are no longer running
- val blocksToDeleteByShuffleService =
- new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
if (externalShuffleServiceRemoveShuffleEnabled) {
- mapOutputTracker.shuffleStatuses.get(shuffleId).foreach { shuffleStatus =>
- shuffleStatus.withMapStatuses { mapStatuses =>
- mapStatuses.foreach { mapStatus =>
- // Check if the executor has been deallocated
- if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
- val blocksToDel =
- shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId, mapStatus.mapId)
- if (blocksToDel.nonEmpty) {
- val blocks = blocksToDeleteByShuffleService.getOrElseUpdate(mapStatus.location,
- new mutable.HashSet[BlockId])
- blocks ++= blocksToDel
+ val shuffleClient = externalBlockStoreClient.get
+ // Find all shuffle blocks on executors that are no longer running
+ val blocksToDelete = new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ mapOutputTracker.shuffleStatuses.get(shuffleId) match {
+ case Some(shuffleStatus) =>
+ shuffleStatus.withMapStatuses { mapStatuses =>
+ mapStatuses.foreach { mapStatus =>
+ // Check if the executor has been deallocated
+ if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
+ val blocksToDel = shuffleManager.shuffleBlockResolver
+ .getBlocksForShuffle(shuffleId, mapStatus.mapId)
+ if (blocksToDel.nonEmpty) {
+ val blocks = blocksToDelete.getOrElseUpdate(mapStatus.location,
+ new mutable.HashSet[BlockId])
+ blocks ++= blocksToDel
+ }
}
}
}
- }
+ removeShuffleFromShuffleServicesFutures ++= blocksToDelete.map { case (bmId, blockIds) =>
+ Future[Boolean] {
+ val numRemovedBlocks = shuffleClient.removeBlocks(
+ bmId.host,
+ bmId.port,
+ bmId.executorId,
+ blockIds.map(_.toString).toArray)
+ numRemovedBlocks.get(defaultRpcTimeout.duration.toSeconds,
+ TimeUnit.SECONDS) == blockIds.size
+ }
+ }.toSeq
+ case None =>
+ logDebug(s"Asked to remove shuffle blocks from " +
+ s"shuffle service for unknown shuffle ${shuffleId}")
}
}
-
- val removeShuffleFromShuffleServicesFutures =
- externalBlockStoreClient.map { shuffleClient =>
- blocksToDeleteByShuffleService.map { case (bmId, blockIds) =>
- Future[Boolean] {
- val numRemovedBlocks = shuffleClient.removeBlocks(
- bmId.host,
- bmId.port,
- bmId.executorId,
- blockIds.map(_.toString).toArray)
- numRemovedBlocks.get(defaultRpcTimeout.duration.toSeconds,
- TimeUnit.SECONDS) == blockIds.size
- }
- }
- }.getOrElse(Seq.empty)
-
+ // It needs to be invoked at last to avoid cleaning up shuffleStatuses in mapOutputTracker
+ // too early before used by [[removeShuffleFromShuffleServicesFutures]]
+ val removeMsg = RemoveShuffle(shuffleId)
+ val removeShuffleFromExecutorsFutures = blockManagerInfo.values.map { bm =>
+ bm.storageEndpoint.ask[Boolean](removeMsg).recover {
+ // use false as default value means no shuffle data were removed
+ handleBlockRemovalFailure("shuffle", shuffleId.toString, bm.blockManagerId, false)
Review Comment:
I just move `removeShuffleFromExecutorsFutures` to the last.
It needs to be invoked at last to avoid cleaning up shuffleStatuses in mapOutputTracker too early. Otherwise `mapOutputTracker.shuffleStatuses.get(shuffleId)` may be none sometimes
Please refer to [unregisterShuffle codes](https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/storage/BlockManagerStorageEndpoint.scala#L59)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wankunde commented on pull request #38560: [SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
wankunde commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1356030075
Hi, @yabola @mridulm , I will update SPARK-40480 this weekend.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1310753366
This is related quite a lot to https://github.com/apache/spark/pull/37922 by @wankunde
That PR seems to be having build issues, and so has not made progress.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1321548579
> One things that I know need to be addressed are:
> Some merge data infos are not saved on the driver because they are too small ( controlled by spark.shuffle.push.minShuffleSizeToWait)
please see https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2295
In this case, we should fire a remove immediately - we are not going to use it for this app anyway ...
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1321551358
@yabola, there is quite a lot of nontrivial overlap between this PR and @wankunde's PR is trying doing.
Would be great if you both can coordinate on this - I would love to get this functionality merged before we starting getting closer to code freeze for 3.4
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1023816561
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -654,8 +731,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
// If no blocks pushed for the finalizeShuffleMerge shuffleMergeId then return
// empty MergeStatuses but cleanup the older shuffleMergeId files.
submitCleanupTask(() ->
- closeAndDeleteOutdatedPartitions(
- appAttemptShuffleMergeId, mergePartitionsInfo.shuffleMergePartitions));
+ deleteCurrentShufflePartitions(appShuffleInfo, shuffleId, appAttemptShuffleMergeId));
Review Comment:
I checked the `appAttemptShuffleMergeId` in the code before.
I think if we want to delete partitions merged data, then we should delete the corresponding ShuffleMergeId in DB (Otherwise, inconsistency will occur when restoring shuffle info from db)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1318238822
I will try to get to this later this week, do let me know if you are still working on it though.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1347915540
> Apologies for the delay in getting to this @yabola - will try to get to this next week. Thanks for your patience.
>
> My recommandation would be to keep the change as close to @wankunde's PR as possible, and fix the pending issues there to expedite the reviews (since that change was already reviewed quite a lot). If there are additional functional gaps in that PR, we can do it in follow up PR's.
@wankunde according to comments, can you fix the remaining comments in your 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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wankunde commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
wankunde commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1029992768
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -452,22 +489,69 @@ void removeAppShuffleInfoFromDB(AppShuffleInfo appShuffleInfo) {
}
/**
- * Clean up all the AppShufflePartitionInfo and the finalized shuffle partitions in DB for
- * a specific shuffleMergeId. This is done since there is a higher shuffleMergeId request made
- * for a shuffleId, therefore clean up older shuffleMergeId partitions. The cleanup will be
- * executed the mergedShuffleCleaner thread.
+ * Clean up the outdated finalized or unfinalized shuffle partitions. The cleanup will be executed in the
+ * mergedShuffleCleaner thread. Two cases need to clean up the shuffle: 1. there is a higher shuffleMergeId request
+ * made for a shuffleId, therefore clean up older shuffleMergeId partitions. 2. Application requires to clean up the
+ * expired or unused specific shuffleId partitions
*/
+ @VisibleForTesting
+ void submitRemoveShuffleMergeTask(
+ AppShuffleInfo shuffleInfo, Integer shuffleId,
+ Optional<AppAttemptShuffleMergeId> higherShuffleMergeId) {
+ AppShuffleMergePartitionsInfo mergePartitionsInfo = shuffleInfo.shuffles.get(shuffleId);
+ AppAttemptShuffleMergeId shuffleMergeId = new AppAttemptShuffleMergeId(
+ shuffleInfo.appId, shuffleInfo.attemptId, shuffleId, mergePartitionsInfo.shuffleMergeId);
+ if (!mergePartitionsInfo.isFinalized()) {
+ Map<Integer, AppShufflePartitionInfo> partitionsToClean = mergePartitionsInfo.shuffleMergePartitions;
+ submitCleanupTask(() ->
+ closeAndDeleteOutdatedPartitions(shuffleMergeId, partitionsToClean, higherShuffleMergeId));
+ } else {
+ int[] partitionsToClean = mergePartitionsInfo.finalizedPartitions;
+ submitCleanupTask(() ->
+ deleteOutdatedFinalizedPartitions(shuffleInfo, shuffleMergeId, partitionsToClean, higherShuffleMergeId));
+ }
+ }
+
@VisibleForTesting
void closeAndDeleteOutdatedPartitions(
- AppAttemptShuffleMergeId appAttemptShuffleMergeId,
- Map<Integer, AppShufflePartitionInfo> partitions) {
- removeAppShufflePartitionInfoFromDB(appAttemptShuffleMergeId);
+ AppAttemptShuffleMergeId shuffleMergeId,
+ Map<Integer, AppShufflePartitionInfo> partitions,
+ Optional<AppAttemptShuffleMergeId> higherShuffleMergeId) {
+ removeAppShufflePartitionInfoFromDB(shuffleMergeId);
+ higherShuffleMergeId.ifPresent(this::removeAppShufflePartitionInfoFromDB);
partitions
- .forEach((partitionId, partitionInfo) -> {
- synchronized (partitionInfo) {
- partitionInfo.closeAllFilesAndDeleteIfNeeded(true);
- }
- });
+ .forEach((partitionId, partitionInfo) -> {
+ synchronized (partitionInfo) {
+ partitionInfo.closeAllFilesAndDeleteIfNeeded(true);
+ }
+ });
+ }
+
+ @VisibleForTesting
+ void deleteOutdatedFinalizedPartitions(
+ AppShuffleInfo shuffleInfo,
+ AppAttemptShuffleMergeId shuffleMergeId,
+ int[] outdatedFinalizedPartitions,
+ Optional<AppAttemptShuffleMergeId> higherShuffleMergeId) {
+ int shuffleId = shuffleMergeId.shuffleId;
+ int mergeId = shuffleMergeId.shuffleMergeId;
+ removeAppShufflePartitionInfoFromDB(shuffleMergeId);
+ higherShuffleMergeId.ifPresent(this::removeAppShufflePartitionInfoFromDB);
+ Arrays.stream(outdatedFinalizedPartitions).forEach(partition -> {
+ try {
+ File metaFile =
+ shuffleInfo.getMergedShuffleMetaFile(shuffleId, mergeId, partition);
+ File indexFile = new File(
+ shuffleInfo.getMergedShuffleIndexFilePath(shuffleId, mergeId, partition));
+ File dataFile =
+ shuffleInfo.getMergedShuffleDataFile(shuffleId, mergeId, partition);
+ metaFile.delete();
+ indexFile.delete();
+ dataFile.delete();
+ } catch (Exception e) {
+ logger.error("Error delete shuffle files for {}", shuffleMergeId, e);
+ }
Review Comment:
Just like `closeAllFilesAndDeleteIfNeeded` method, can we continue delete the other files if one `delete()` failed?
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -712,6 +794,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
mergeStatuses = new MergeStatuses(msg.shuffleId, msg.shuffleMergeId,
bitmaps.toArray(new RoaringBitmap[bitmaps.size()]), Ints.toArray(reduceIds),
Longs.toArray(sizes));
+ appShuffleInfo.shuffles.get(msg.shuffleId).setFinalizedPartitions(Ints.toArray(reduceIds));
Review Comment:
The `FinalizedPartitions` will be empty after the Shuffle service restart which will cause the merged shuffle files leak.
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -327,43 +327,52 @@ class BlockManagerMasterEndpoint(
}
}.toSeq
- // Find all shuffle blocks on executors that are no longer running
- val blocksToDeleteByShuffleService =
- new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ var removeShuffleFromShuffleServicesFutures = Seq.empty[Future[Boolean]]
if (externalShuffleServiceRemoveShuffleEnabled) {
- mapOutputTracker.shuffleStatuses.get(shuffleId).foreach { shuffleStatus =>
- shuffleStatus.withMapStatuses { mapStatuses =>
- mapStatuses.foreach { mapStatus =>
- // Check if the executor has been deallocated
- if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
- val blocksToDel =
- shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId, mapStatus.mapId)
- if (blocksToDel.nonEmpty) {
- val blocks = blocksToDeleteByShuffleService.getOrElseUpdate(mapStatus.location,
- new mutable.HashSet[BlockId])
- blocks ++= blocksToDel
+ val shuffleClient = externalBlockStoreClient.get
+ // Find all shuffle blocks on executors that are no longer running
+ val blocksToDelete = new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ mapOutputTracker.shuffleStatuses.get(shuffleId) match {
+ case Some(shuffleStatus) =>
+ shuffleStatus.withMapStatuses { mapStatuses =>
+ mapStatuses.foreach { mapStatus =>
+ // Check if the executor has been deallocated
+ if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
+ val blocksToDel = shuffleManager.shuffleBlockResolver
+ .getBlocksForShuffle(shuffleId, mapStatus.mapId)
+ if (blocksToDel.nonEmpty) {
+ val blocks = blocksToDelete.getOrElseUpdate(mapStatus.location,
+ new mutable.HashSet[BlockId])
+ blocks ++= blocksToDel
+ }
Review Comment:
What if the shuffle statuses are not exists ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1031346915
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -319,51 +319,68 @@ class BlockManagerMasterEndpoint(
}
private def removeShuffle(shuffleId: Int): Future[Seq[Boolean]] = {
- val removeMsg = RemoveShuffle(shuffleId)
- val removeShuffleFromExecutorsFutures = blockManagerInfo.values.map { bm =>
- bm.storageEndpoint.ask[Boolean](removeMsg).recover {
- // use false as default value means no shuffle data were removed
- handleBlockRemovalFailure("shuffle", shuffleId.toString, bm.blockManagerId, false)
+ var removeShuffleFromShuffleServicesFutures = Seq.empty[Future[Boolean]]
+ if (pushBasedShuffleEnabled && externalBlockStoreClient.isDefined) {
+ val shuffleClient = externalBlockStoreClient.get
+ mapOutputTracker.shuffleStatuses.get(shuffleId) match {
+ case Some(shuffleStatus) =>
+ val shuffleMergerLocations = shuffleStatus.getShufflePushMergerLocations
+ removeShuffleFromShuffleServicesFutures ++= shuffleMergerLocations.map(bmId => {
+ Future[Boolean] {
+ shuffleClient.removeShuffleMerge(bmId.host, bmId.port, shuffleId)
+ }
+ })
+ case None =>
+ logWarning(s"Asked to remove merge shuffle blocks from " +
+ s"shuffle service for unknown shuffle ${shuffleId}")
}
- }.toSeq
+ }
- // Find all shuffle blocks on executors that are no longer running
- val blocksToDeleteByShuffleService =
- new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
if (externalShuffleServiceRemoveShuffleEnabled) {
- mapOutputTracker.shuffleStatuses.get(shuffleId).foreach { shuffleStatus =>
- shuffleStatus.withMapStatuses { mapStatuses =>
- mapStatuses.foreach { mapStatus =>
- // Check if the executor has been deallocated
- if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
- val blocksToDel =
- shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId, mapStatus.mapId)
- if (blocksToDel.nonEmpty) {
- val blocks = blocksToDeleteByShuffleService.getOrElseUpdate(mapStatus.location,
- new mutable.HashSet[BlockId])
- blocks ++= blocksToDel
+ val shuffleClient = externalBlockStoreClient.get
+ // Find all shuffle blocks on executors that are no longer running
+ val blocksToDelete = new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
+ mapOutputTracker.shuffleStatuses.get(shuffleId) match {
+ case Some(shuffleStatus) =>
+ shuffleStatus.withMapStatuses { mapStatuses =>
+ mapStatuses.foreach { mapStatus =>
+ // Check if the executor has been deallocated
+ if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
+ val blocksToDel = shuffleManager.shuffleBlockResolver
+ .getBlocksForShuffle(shuffleId, mapStatus.mapId)
+ if (blocksToDel.nonEmpty) {
+ val blocks = blocksToDelete.getOrElseUpdate(mapStatus.location,
+ new mutable.HashSet[BlockId])
+ blocks ++= blocksToDel
+ }
}
}
}
- }
+ removeShuffleFromShuffleServicesFutures ++= blocksToDelete.map { case (bmId, blockIds) =>
+ Future[Boolean] {
+ val numRemovedBlocks = shuffleClient.removeBlocks(
+ bmId.host,
+ bmId.port,
+ bmId.executorId,
+ blockIds.map(_.toString).toArray)
+ numRemovedBlocks.get(defaultRpcTimeout.duration.toSeconds,
+ TimeUnit.SECONDS) == blockIds.size
+ }
+ }.toSeq
+ case None =>
+ logDebug(s"Asked to remove shuffle blocks from " +
+ s"shuffle service for unknown shuffle ${shuffleId}")
}
}
-
- val removeShuffleFromShuffleServicesFutures =
- externalBlockStoreClient.map { shuffleClient =>
- blocksToDeleteByShuffleService.map { case (bmId, blockIds) =>
- Future[Boolean] {
- val numRemovedBlocks = shuffleClient.removeBlocks(
- bmId.host,
- bmId.port,
- bmId.executorId,
- blockIds.map(_.toString).toArray)
- numRemovedBlocks.get(defaultRpcTimeout.duration.toSeconds,
- TimeUnit.SECONDS) == blockIds.size
- }
- }
- }.getOrElse(Seq.empty)
-
+ // It needs to be invoked at last to avoid cleaning up shuffleStatuses in mapOutputTracker
+ // too early before used by [[removeShuffleFromShuffleServicesFutures]]
+ val removeMsg = RemoveShuffle(shuffleId)
+ val removeShuffleFromExecutorsFutures = blockManagerInfo.values.map { bm =>
+ bm.storageEndpoint.ask[Boolean](removeMsg).recover {
+ // use false as default value means no shuffle data were removed
+ handleBlockRemovalFailure("shuffle", shuffleId.toString, bm.blockManagerId, false)
Review Comment:
I just move `removeShuffleFromExecutorsFutures` to the last.
It needs to be invoked at last to avoid cleaning up shuffleStatuses in mapOutputTracker to early. Otherwise `mapOutputTracker.shuffleStatuses.get(shuffleId)` may be none sometimes
Please refer to [unregisterShuffle codes](https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/storage/BlockManagerStorageEndpoint.scala#L59)
##########
resource-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceSuite.scala:
##########
@@ -956,7 +956,7 @@ abstract class YarnShuffleServiceSuite extends SparkFunSuite with Matchers {
assert(ShuffleTestAccessor.getOutdatedAppPathInfoCountDuringDBReload(
mergeManager2, mergeManager2DB) == 1)
assert(ShuffleTestAccessor.getOutdatedFinalizedShuffleCountDuringDBReload(
- mergeManager2, mergeManager2DB) == 2)
+ mergeManager2, mergeManager2DB) == 1)
Review Comment:
This is as expected, because we delete the current merge partitions and the current outdated merge status in db (not cleaned before this PR)
Please refer to [codes](https://github.com/apache/spark/pull/38560/files#diff-d544fbb952b61283b3d18ca10a5027528efc4f2f65047130da015ae7754c117fR506)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1333302610
@mridulm If you are back and have time, please review my PR, I think the function is almost done. let me know if there is something inappropriate, I will modify it soon, thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1023828427
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -654,8 +731,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
// If no blocks pushed for the finalizeShuffleMerge shuffleMergeId then return
// empty MergeStatuses but cleanup the older shuffleMergeId files.
submitCleanupTask(() ->
- closeAndDeleteOutdatedPartitions(
- appAttemptShuffleMergeId, mergePartitionsInfo.shuffleMergePartitions));
+ deleteCurrentShufflePartitions(appShuffleInfo, shuffleId, appAttemptShuffleMergeId));
Review Comment:
And I think when deleting partitions, we shouldn't store shuffleMergeID in DB anymore
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1311193090
my latest implementation no longer passes reduceIds from driver, there are still some code style improvements, just some rough implementation for now
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1023828427
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -654,8 +731,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
// If no blocks pushed for the finalizeShuffleMerge shuffleMergeId then return
// empty MergeStatuses but cleanup the older shuffleMergeId files.
submitCleanupTask(() ->
- closeAndDeleteOutdatedPartitions(
- appAttemptShuffleMergeId, mergePartitionsInfo.shuffleMergePartitions));
+ deleteCurrentShufflePartitions(appShuffleInfo, shuffleId, appAttemptShuffleMergeId));
Review Comment:
And I think when deleting partition information, we shouldn't store shuffleMergeID in DB anymore
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1023816561
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -654,8 +731,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
// If no blocks pushed for the finalizeShuffleMerge shuffleMergeId then return
// empty MergeStatuses but cleanup the older shuffleMergeId files.
submitCleanupTask(() ->
- closeAndDeleteOutdatedPartitions(
- appAttemptShuffleMergeId, mergePartitionsInfo.shuffleMergePartitions));
+ deleteCurrentShufflePartitions(appShuffleInfo, shuffleId, appAttemptShuffleMergeId));
Review Comment:
~~I checked the `appAttemptShuffleMergeId` in the code before.
I think if we want to delete partitions merged data, then we should delete the corresponding ShuffleMergeId in DB (Otherwise, inconsistency will occur when restoring shuffle info from db)~~
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola commented on a diff in pull request #38560: [WIP][SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola commented on code in PR #38560:
URL: https://github.com/apache/spark/pull/38560#discussion_r1030395732
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -712,6 +794,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) {
mergeStatuses = new MergeStatuses(msg.shuffleId, msg.shuffleMergeId,
bitmaps.toArray(new RoaringBitmap[bitmaps.size()]), Ints.toArray(reduceIds),
Longs.toArray(sizes));
+ appShuffleInfo.shuffles.get(msg.shuffleId).setFinalizedPartitions(Ints.toArray(reduceIds));
Review Comment:
What if the shuffle statuses are not exists ?
I think it will not, please see [case match codes](https://github.com/apache/spark/pull/38560/files#diff-21bb40987f7d21c8d3d3de3212f388c210fb04a636c96bdef53cab6070c6aa46R336)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #38560: [SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38560:
URL: https://github.com/apache/spark/pull/38560#issuecomment-1336104208
Apologies for the delay in getting to this @yabola - will try to get to this next week.
Thanks for your patience.
My recommandation would be to keep the change as close to @wankunde's PR as possible, and fix the pending issues there to expedite the reviews (since that change was already reviewed quite a lot). If there are additional gaps in that PR, we can do it in follow up PR's.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] yabola closed pull request #38560: [SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
Posted by GitBox <gi...@apache.org>.
yabola closed pull request #38560: [SPARK-38005][core] Support cleaning up merged shuffle files and state from external shuffle service
URL: https://github.com/apache/spark/pull/38560
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org