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 2021/12/10 14:55:46 UTC
[GitHub] [spark] tgravescs commented on a change in pull request #34725: [SPARK-37473][CORE] BypassMergeSortShuffleWriter may loss data when disk is missing however catagory is present
tgravescs commented on a change in pull request #34725:
URL: https://github.com/apache/spark/pull/34725#discussion_r766740700
##########
File path: core/src/main/java/org/apache/spark/shuffle/sort/BypassMergeSortShuffleWriter.java
##########
@@ -201,29 +206,36 @@ public void write(Iterator<Product2<K, V>> records) throws IOException {
*
* @return array of lengths, in bytes, of each partition of the file (used by map output tracker).
*/
- private long[] writePartitionedData(ShuffleMapOutputWriter mapOutputWriter) throws IOException {
+ private long[] writePartitionedData(
+ ShuffleMapOutputWriter mapOutputWriter,
+ Set<Integer> nonEmptyPartitionSet) throws IOException {
Review comment:
should add description of parameters and why we need the nonEmptyPartitionSet
##########
File path: core/src/main/java/org/apache/spark/shuffle/sort/BypassMergeSortShuffleWriter.java
##########
@@ -201,29 +206,36 @@ public void write(Iterator<Product2<K, V>> records) throws IOException {
*
* @return array of lengths, in bytes, of each partition of the file (used by map output tracker).
*/
- private long[] writePartitionedData(ShuffleMapOutputWriter mapOutputWriter) throws IOException {
+ private long[] writePartitionedData(
+ ShuffleMapOutputWriter mapOutputWriter,
+ Set<Integer> nonEmptyPartitionSet) throws IOException {
// Track location of the partition starts in the output file
if (partitionWriters != null) {
final long writeStartTime = System.nanoTime();
try {
for (int i = 0; i < numPartitions; i++) {
- final File file = partitionWriterSegments[i].file();
- ShufflePartitionWriter writer = mapOutputWriter.getPartitionWriter(i);
- if (file.exists()) {
- if (transferToEnabled) {
- // Using WritableByteChannelWrapper to make resource closing consistent between
- // this implementation and UnsafeShuffleWriter.
- Optional<WritableByteChannelWrapper> maybeOutputChannel = writer.openChannelWrapper();
- if (maybeOutputChannel.isPresent()) {
- writePartitionedDataWithChannel(file, maybeOutputChannel.get());
+ if (nonEmptyPartitionSet.contains(i)) {
+ final File file = partitionWriterSegments[i].file();
+ ShufflePartitionWriter writer = mapOutputWriter.getPartitionWriter(i);
+ if (file.exists()) {
+ if (transferToEnabled) {
+ // Using WritableByteChannelWrapper to make resource closing consistent between
+ // this implementation and UnsafeShuffleWriter.
+ Optional<WritableByteChannelWrapper> maybeOutputChannel = writer.openChannelWrapper();
+ if (maybeOutputChannel.isPresent()) {
+ writePartitionedDataWithChannel(file, maybeOutputChannel.get());
+ } else {
+ writePartitionedDataWithStream(file, writer);
+ }
} else {
writePartitionedDataWithStream(file, writer);
}
+ if (!file.delete()) {
+ logger.error("Unable to delete file for partition {}", i);
+ }
} else {
- writePartitionedDataWithStream(file, writer);
- }
- if (!file.delete()) {
- logger.error("Unable to delete file for partition {}", i);
+ throw new IOException(
+ "Segment file " + file.getAbsolutePath() + " for partition " + i + " doesn't exists.");
Review comment:
might be nice to add more description to the exception for user to understand what might have happened
--
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