Skip to content

[SPARK-8029] Robust shuffle writer #9610

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Closed
davies wants to merge 8 commits into from
Closed

[SPARK-8029] Robust shuffle writer #9610

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
b2a90c6
use temporary file and rename to avoid concurrent write
Nov 10, 2015
9f0d2f9
fix tests
Nov 11, 2015
55485a9
address comments
Nov 11, 2015
6deccff
fix test
Nov 11, 2015
f0c2a5d
address comments
Nov 12, 2015
d0b937f
rename
Nov 12, 2015
35bd469
add regression test
Nov 13, 2015
71b12bf
address comments
Nov 13, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions core/src/main/java/org/apache/spark/shuffle/sort/BypassMergeSortShuffleWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.UUID;
import javax.annotation.Nullable;

import scala.None$;
Expand Down Expand Up @@ -155,9 +156,20 @@ public void write(Iterator<Product2<K, V>> records) throws IOException {
writer.commitAndClose();
}

partitionLengths =
writePartitionedFile(shuffleBlockResolver.getDataFile(shuffleId, mapId));
shuffleBlockResolver.writeIndexFile(shuffleId, mapId, partitionLengths);
File output = shuffleBlockResolver.getDataFile(shuffleId, mapId);
final File tmp = new File(output.getAbsolutePath() + "." + UUID.randomUUID());

@squito squito Nov 11, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

good point about creating the tmp files in the same dir as the dest to make sure we can do the rename ... I had taken that for granted.

partitionLengths = writePartitionedFile(tmp);
if (!output.exists()) {

@squito squito Nov 11, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I dont' think you can do this and still support SPARK-4085 -- regenerating the output if one of the shuffle files goes completely missing. Because if the index file goes missing, and the data file is still there, with this logic you'll always never regenerate the shuffle output. But maybe SPARK-4085 is not worth it ...

@davies davies Nov 11, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Good point, we should check both.

shuffleBlockResolver.writeIndexFile(shuffleId, mapId, partitionLengths);
if (output.exists()) {
output.delete();
}
if (!tmp.renameTo(output)) {
throw new IOException("fail to rename data file " + tmp + " to " + output);
}
} else {
tmp.delete();
}
mapStatus = MapStatus$.MODULE$.apply(blockManager.shuffleServerId(), partitionLengths);
}

Expand Down
23 changes: 17 additions & 6 deletions core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import java.io.*;
import java.nio.channels.FileChannel;
import java.util.Iterator;
import java.util.UUID;

import scala.Option;
import scala.Product2;
Expand All @@ -41,7 +42,7 @@
import org.apache.spark.executor.ShuffleWriteMetrics;
import org.apache.spark.io.CompressionCodec;
import org.apache.spark.io.CompressionCodec$;
import org.apache.spark.io.LZFCompressionCodec;
import org.apache.spark.memory.TaskMemoryManager;
import org.apache.spark.network.util.LimitedInputStream;
import org.apache.spark.scheduler.MapStatus;
import org.apache.spark.scheduler.MapStatus$;
Expand All @@ -53,7 +54,6 @@
import org.apache.spark.storage.BlockManager;
import org.apache.spark.storage.TimeTrackingOutputStream;
import org.apache.spark.unsafe.Platform;
import org.apache.spark.memory.TaskMemoryManager;

@Private
public class UnsafeShuffleWriter<K, V> extends ShuffleWriter<K, V> {
Expand Down Expand Up @@ -206,16 +206,28 @@ void closeAndWriteOutput() throws IOException {
final SpillInfo[] spills = sorter.closeAndGetSpills();
sorter = null;
final long[] partitionLengths;
final File output = shuffleBlockResolver.getDataFile(shuffleId, mapId);
final File tmp = new File(output.getAbsolutePath() + "." + UUID.randomUUID());
try {
partitionLengths = mergeSpills(spills);
partitionLengths = mergeSpills(spills, tmp);
} finally {
for (SpillInfo spill : spills) {
if (spill.file.exists() && ! spill.file.delete()) {
logger.error("Error while deleting spill file {}", spill.file.getPath());
}
}
}
shuffleBlockResolver.writeIndexFile(shuffleId, mapId, partitionLengths);
if (!output.exists()) {
shuffleBlockResolver.writeIndexFile(shuffleId, mapId, partitionLengths);
if (output.exists()) {
output.delete();
}
if (!tmp.renameTo(output)) {
throw new IOException("fail to rename data file " + tmp + " to " + output);
}
} else {
tmp.delete();
}
mapStatus = MapStatus$.MODULE$.apply(blockManager.shuffleServerId(), partitionLengths);
}

Expand Down Expand Up @@ -248,8 +260,7 @@ void forceSorterToSpill() throws IOException {
*
* @return the partition lengths in the merged file.
*/
private long[] mergeSpills(SpillInfo[] spills) throws IOException {
final File outputFile = shuffleBlockResolver.getDataFile(shuffleId, mapId);
private long[] mergeSpills(SpillInfo[] spills, File outputFile) throws IOException {
final boolean compressionEnabled = sparkConf.getBoolean("spark.shuffle.compress", true);
final CompressionCodec compressionCodec = CompressionCodec$.MODULE$.createCodec(sparkConf);
final boolean fastMergeEnabled =
Expand Down
15 changes: 4 additions & 11 deletions core/src/main/scala/org/apache/spark/shuffle/FileShuffleBlockResolver.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.spark.shuffle

import java.io.File
import java.util.UUID
import java.util.concurrent.ConcurrentLinkedQueue

import scala.collection.JavaConverters._
Expand Down Expand Up @@ -84,17 +86,8 @@ private[spark] class FileShuffleBlockResolver(conf: SparkConf)
Array.tabulate[DiskBlockObjectWriter](numReducers) { bucketId =>
val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
val blockFile = blockManager.diskBlockManager.getFile(blockId)
// Because of previous failures, the shuffle file may already exist on this machine.
// If so, remove it.
if (blockFile.exists) {
if (blockFile.delete()) {
logInfo(s"Removed existing shuffle file $blockFile")
} else {
logWarning(s"Failed to remove existing shuffle file $blockFile")
}
}
blockManager.getDiskWriter(blockId, blockFile, serializerInstance, bufferSize,
writeMetrics)
val tmp = new File(blockFile.getAbsolutePath + "." + UUID.randomUUID())

@andrewor14 andrewor14 Nov 13, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

why not use your new method Utils.withTempFile

blockManager.getDiskWriter(blockId, tmp, serializerInstance, bufferSize, writeMetrics)
}
}
// Creating the file to write to and creating a disk writer both involve interacting with
Expand Down
8 changes: 7 additions & 1 deletion core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.shuffle

import java.io._
import java.util.UUID

import com.google.common.io.ByteStreams

Expand Down Expand Up @@ -81,7 +82,8 @@ private[spark] class IndexShuffleBlockResolver(conf: SparkConf) extends ShuffleB
* */
def writeIndexFile(shuffleId: Int, mapId: Int, lengths: Array[Long]): Unit = {
val indexFile = getIndexFile(shuffleId, mapId)
val out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(indexFile)))
val tmp = new File(indexFile.getAbsolutePath + "." + UUID.randomUUID())
val out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(tmp)))
Utils.tryWithSafeFinally {
// We take in lengths of each block, need to convert it to offsets.
var offset = 0L
Expand All @@ -93,6 +95,10 @@ private[spark] class IndexShuffleBlockResolver(conf: SparkConf) extends ShuffleB
} {
out.close()
}
indexFile.deleteOnExit()
if (!tmp.renameTo(indexFile)) {
throw new IOException(s"fail to rename index file $tmp to $indexFile")

@squito squito Nov 11, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

this will just kill the task, right? both tasks are actually just fine, and in fact the overall job should continue if one of them succeeds. But instead this will lead to the task getting retried, and potentially continuing to fail up to 4 times, though its actually finished successfully from another taskset? You could handle this in scheduler, but that would add some complexity.

@davies davies Nov 11, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

There is very little chance that the two concurrent task will call renameTo in the same time, even with that, one of them will succeed, the scheduler will mark the partition as success, and the failure will be ignored (not retried).

@mateiz mateiz Nov 12, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Can you test for this? I think the worry was about different TaskSets attempting the same map stage. Imagine that attempt 1 of the stage successfully completes a task, and sends back a map output status, but that status gets ignored because that stage attempt got cancelled. Attempt 2 might then fail to send a new status for it.

There seem to be two ways to fix it if this problem can actually occur -- either add MapOutputStatuses even from failed task sets or mark this new task as successful if a file exists.

@davies davies Nov 12, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

On Thu, Nov 12, 2015 at 8:30 AM, Matei Zaharia notifications@github.com
wrote:

In
core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala
#9610 (comment):

@@ -93,6 +95,10 @@ private[spark] class IndexShuffleBlockResolver(conf: SparkConf) extends ShuffleB
} {
out.close()
}

  • indexFile.deleteOnExit()
  • if (!tmp.renameTo(indexFile)) {
  •  throw new IOException(s"fail to rename index file $tmp to $indexFile")

Can you test for this? I think the worry was about different TaskSets
attempting the same map stage. Imagine that attempt 1 of the stage
successfully completes a task, and sends back a map output status, but that
status gets ignored because that stage attempt got cancelled. Attempt 2
might then fail to send a new status for it.

There seem to be two ways to fix it if this problem can actually occur --
either add MapOutputStatuses even from failed task sets or mark this new
task as successful if a file exists.

After this PR, the second attempt of same task will return SUCCESS, with
new MapOutputStatus, which could be different than the previous attempt
(having different sizes of partitions), since we does not use the exact
number of size (could be lossy compressed), I think it's fine.

—
Reply to this email directly or view it on GitHub
https://github.com/apache/spark/pull/9610/files#r44678796.

  • Davies

}
}

override def getBlockData(blockId: ShuffleBlockId): ManagedBuffer = {
Expand Down
15 changes: 15 additions & 0 deletions core/src/main/scala/org/apache/spark/shuffle/hash/HashShuffleWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.spark.shuffle.hash

import java.io.IOException

import org.apache.spark._
import org.apache.spark.executor.ShuffleWriteMetrics
import org.apache.spark.scheduler.MapStatus
Expand Down Expand Up @@ -106,6 +108,19 @@ private[spark] class HashShuffleWriter[K, V](
writer.commitAndClose()
writer.fileSegment().length
}
// rename all shuffle files to final paths
shuffle.writers.zip(sizes).foreach { case (writer: DiskBlockObjectWriter, size: Long) =>
if (size > 0) {
val output = blockManager.diskBlockManager.getFile(writer.blockId)
if (output.exists()) {
writer.file.delete()
} else {
if (!writer.file.renameTo(output)) {
throw new IOException(s"fail to rename ${writer.file} to $output")

@squito squito Nov 11, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

same problem here on partially existing shuffle output. Also, the if (size > 0) check will lead to inconsistencies if you have non-deterministic shuffle output -- it might be true for different partitions in different attempts. I think it can be OK in any case, as long as the MapStatus is consistent.

@davies davies Nov 11, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I think these partitions are independent, they should be OK whenever it's generated in different attempt, or that's the basic idea of how RDD works (could be re-run and got the same result). If not, for example, the items in RDD is random are generated randomly, then it also does not matter if it's different across attempts.

@davies davies Nov 11, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

The size here is used to make sure that a new file is generated (we could try to delete existed file if size is 0).

@squito squito Nov 12, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

yeah I suppose it all depends on what the model is for non-deterministic data. The reduce tasks can read data from a mix attempts, but I guess that is OK (we can't completely prevent it in any case). There is also the problem of returning the right mapstatus here, but it doesn't matter as much in this case -- you will at least return some set of non-empty blocks that is consistent with the shuffle data on disk, even if the sizes can be arbitrarily wrong.

Also I know its super-rare, but there is a race between output.exists and renameTo(output), might as well protect against that.

I also find it a weird that this is neither first or last attempt wins -- the first attempt to get to each output file wins, but it can be a mix of attempts. again I'd include a comment explaining the logic

}
}
}
}
MapStatus(blockManager.shuffleServerId, sizes)
}

Expand Down
20 changes: 17 additions & 3 deletions core/src/main/scala/org/apache/spark/shuffle/sort/SortShuffleWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@

package org.apache.spark.shuffle.sort

import java.io.{IOException, File}
import java.util.UUID

import org.apache.spark._
import org.apache.spark.executor.ShuffleWriteMetrics
import org.apache.spark.scheduler.MapStatus
Expand Down Expand Up @@ -65,10 +68,21 @@ private[spark] class SortShuffleWriter[K, V, C](
// Don't bother including the time to open the merged output file in the shuffle write time,
// because it just opens a single file, so is typically too fast to measure accurately
// (see SPARK-3570).
val outputFile = shuffleBlockResolver.getDataFile(dep.shuffleId, mapId)
val output = shuffleBlockResolver.getDataFile(dep.shuffleId, mapId)
val tmp = new File(output.getAbsolutePath + "." + UUID.randomUUID())
val blockId = ShuffleBlockId(dep.shuffleId, mapId, IndexShuffleBlockResolver.NOOP_REDUCE_ID)
val partitionLengths = sorter.writePartitionedFile(blockId, outputFile)
shuffleBlockResolver.writeIndexFile(dep.shuffleId, mapId, partitionLengths)
val partitionLengths = sorter.writePartitionedFile(blockId, tmp)
if (!output.exists()) {
shuffleBlockResolver.writeIndexFile(dep.shuffleId, mapId, partitionLengths)
if (output.exists()) {
output.delete()
}
if (!tmp.renameTo(output)) {
throw new IOException("fail to rename data file " + tmp + " to " + output)
}
} else {
tmp.delete()
}

mapStatus = MapStatus(blockManager.shuffleServerId, partitionLengths)
}
Expand Down
9 changes: 4 additions & 5 deletions core/src/main/scala/org/apache/spark/storage/BlockManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,10 @@ import java.io._
import java.nio.{ByteBuffer, MappedByteBuffer}

import scala.collection.mutable.{ArrayBuffer, HashMap}
import scala.concurrent.{ExecutionContext, Await, Future}
import scala.concurrent.duration._
import scala.util.control.NonFatal
import scala.concurrent.{Await, ExecutionContext, Future}
import scala.util.Random
import scala.util.control.NonFatal

import sun.nio.ch.DirectBuffer

Expand All @@ -38,9 +38,8 @@ import org.apache.spark.network.netty.SparkTransportConf
import org.apache.spark.network.shuffle.ExternalShuffleClient
import org.apache.spark.network.shuffle.protocol.ExecutorShuffleInfo
import org.apache.spark.rpc.RpcEnv
import org.apache.spark.serializer.{SerializerInstance, Serializer}
import org.apache.spark.serializer.{Serializer, SerializerInstance}
import org.apache.spark.shuffle.ShuffleManager
import org.apache.spark.shuffle.hash.HashShuffleManager
import org.apache.spark.util._

private[spark] sealed trait BlockValues
Expand Down Expand Up @@ -660,7 +659,7 @@ private[spark] class BlockManager(
val compressStream: OutputStream => OutputStream = wrapForCompression(blockId, _)
val syncWrites = conf.getBoolean("spark.shuffle.sync", false)
new DiskBlockObjectWriter(file, serializerInstance, bufferSize, compressStream,
syncWrites, writeMetrics)
syncWrites, writeMetrics, blockId)
}

/**
Expand Down
5 changes: 3 additions & 2 deletions core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,14 +34,15 @@ import org.apache.spark.util.Utils
* reopened again.
*/
private[spark] class DiskBlockObjectWriter(
file: File,
val file: File,
serializerInstance: SerializerInstance,
bufferSize: Int,
compressStream: OutputStream => OutputStream,
syncWrites: Boolean,
// These write metrics concurrently shared with other active DiskBlockObjectWriters who
// are themselves performing writes. All updates must be relative.
writeMetrics: ShuffleWriteMetrics)
writeMetrics: ShuffleWriteMetrics,
val blockId: BlockId = null)
extends OutputStream
with Logging {

Expand Down
1 change: 0 additions & 1 deletion core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -639,7 +639,6 @@ private[spark] class ExternalSorter[K, V, C](
* called by the SortShuffleWriter.
*
* @param blockId block ID to write to. The index file will be blockId.name + ".index".
* @param context a TaskContext for a running Spark task, for us to update shuffle metrics.
* @return array of lengths, in bytes, of each partition of the file (used by map output tracker)
*/
def writePartitionedFile(
Expand Down
3 changes: 2 additions & 1 deletion core/src/test/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriterSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,8 @@ public DiskBlockObjectWriter answer(InvocationOnMock invocationOnMock) throws Th
(Integer) args[3],
new CompressStream(),
false,
(ShuffleWriteMetrics) args[4]
(ShuffleWriteMetrics) args[4],
(BlockId) args[0]
);
}
});
Expand Down
3 changes: 2 additions & 1 deletion core/src/test/java/org/apache/spark/unsafe/map/AbstractBytesToBytesMapSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,8 @@ public DiskBlockObjectWriter answer(InvocationOnMock invocationOnMock) throws Th
(Integer) args[3],
new CompressStream(),
false,
(ShuffleWriteMetrics) args[4]
(ShuffleWriteMetrics) args[4],
(BlockId) args[0]
);
}
});
Expand Down
3 changes: 2 additions & 1 deletion ...src/test/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorterSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,8 @@ public DiskBlockObjectWriter answer(InvocationOnMock invocationOnMock) throws Th
(Integer) args[3],
new CompressStream(),
false,
(ShuffleWriteMetrics) args[4]
(ShuffleWriteMetrics) args[4],
(BlockId) args[0]
);
}
});
Expand Down
3 changes: 2 additions & 1 deletion core/src/test/scala/org/apache/spark/shuffle/sort/BypassMergeSortShuffleWriterSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,8 @@ class BypassMergeSortShuffleWriterSuite extends SparkFunSuite with BeforeAndAfte
args(3).asInstanceOf[Int],
compressStream = identity,
syncWrites = false,
args(4).asInstanceOf[ShuffleWriteMetrics]
args(4).asInstanceOf[ShuffleWriteMetrics],
blockId = args(0).asInstanceOf[BlockId]
)
}
})
Expand Down