[SPARK-8029] Robust shuffle writer #9610
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,13 +21,12 @@ import java.io._ | |
|
|
||
| import com.google.common.io.ByteStreams | ||
|
|
||
| import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer} | ||
| import org.apache.spark.network.netty.SparkTransportConf | ||
| import org.apache.spark.shuffle.IndexShuffleBlockResolver.NOOP_REDUCE_ID | ||
| import org.apache.spark.storage._ | ||
| import org.apache.spark.util.Utils | ||
| import org.apache.spark.{SparkEnv, Logging, SparkConf} | ||
|
|
||
| /** | ||
| * Create and maintain the shuffle blocks' mapping between logic block and physical file location. | ||
|
|
@@ -40,10 +39,17 @@ import IndexShuffleBlockResolver.NOOP_REDUCE_ID | |
| */ | ||
| // Note: Changes to the format in this file should be kept in sync with | ||
| // org.apache.spark.network.shuffle.ExternalShuffleBlockResolver#getSortBasedShuffleBlockData(). | ||
| private[spark] class IndexShuffleBlockResolver( | ||
| conf: SparkConf, | ||
| _blockManager: BlockManager = null) | ||
| extends ShuffleBlockResolver | ||
| with Logging { | ||
|
|
||
| private lazy val blockManager = if (_blockManager == null) { | ||
| SparkEnv.get.blockManager | ||
| } else { | ||
| _blockManager | ||
| } | ||
|
There was a problem hiding this comment. or |
||
|
|
||
| private val transportConf = SparkTransportConf.fromSparkConf(conf) | ||
|
|
||
|
|
@@ -74,14 +80,69 @@ private[spark] class IndexShuffleBlockResolver(conf: SparkConf) extends ShuffleB | |
| } | ||
| } | ||
|
|
||
| /** | ||
| * Check whether there are index file and data file also they are matched with each other, returns | ||
| * the lengths of each block in data file, if there are matched, or return null. | ||
|
There was a problem hiding this comment. |
||
| */ | ||
| private def checkIndexAndDataFile(index: File, data: File, blocks: Int): Array[Long] = { | ||
| val lengths = new Array[Long](blocks) | ||
| if (index.length() == (blocks + 1) * 8) { | ||
|
There was a problem hiding this comment. instead of the nested |
||
| // Read the lengths of blocks | ||
| val f = try { | ||
| new FileInputStream(index) | ||
| } catch { | ||
| case e: IOException => | ||
| return null | ||
|
There was a problem hiding this comment. should we at least |
||
| } | ||
| val in = new DataInputStream(new BufferedInputStream(f)) | ||
| try { | ||
| // Convert the offsets into lengths of each block | ||
| var offset = in.readLong() | ||
| if (offset != 0L) { | ||
| return null | ||
| } | ||
| var i = 0 | ||
| while (i < blocks) { | ||
| val off = in.readLong() | ||
| lengths(i) = off - offset | ||
| offset = off | ||
| i += 1 | ||
| } | ||
| } catch { | ||
| case e: IOException => | ||
| return null | ||
| } finally { | ||
| in.close() | ||
| } | ||
|
|
||
| val size = lengths.reduce(_ + _) | ||
| // `length` returns 0 if it not exists. | ||
| if (data.length() == size) { | ||
|
There was a problem hiding this comment. or just |
||
| lengths | ||
| } else { | ||
| null | ||
| } | ||
| } else { | ||
| null | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Write an index file with the offsets of each block, plus a final offset at the end for the | ||
| * end of the output file. This will be used by getBlockData to figure out where each block | ||
| * begins and ends. | ||
| * | ||
| * It will commit the data and index file as an atomic operation, use the existed ones (lengths of | ||
| * blocks will be refreshed), or replace them with new ones. | ||
|
There was a problem hiding this comment. we should add that this modifies the contents of |
||
| * */ | ||
| def writeIndexFileAndCommit( | ||
| shuffleId: Int, | ||
| mapId: Int, | ||
| lengths: Array[Long], | ||
| dataTmp: File): Unit = { | ||
| val indexFile = getIndexFile(shuffleId, mapId) | ||
| val indexTmp = Utils.tempFileWith(indexFile) | ||
| val out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(indexTmp))) | ||
| Utils.tryWithSafeFinally { | ||
| // We take in lengths of each block, need to convert it to offsets. | ||
| var offset = 0L | ||
|
|
@@ -93,6 +154,31 @@ private[spark] class IndexShuffleBlockResolver(conf: SparkConf) extends ShuffleB | |
| } { | ||
| out.close() | ||
| } | ||
|
|
||
| val dataFile = getDataFile(shuffleId, mapId) | ||
| // Note: there is only one IndexShuffleBlockResolver per executor | ||
| synchronized { | ||
|
There was a problem hiding this comment. Can you add a big comment explaining what is going on here? Also worth noting that there is only one There was a problem hiding this comment. is this |
||
| val existedLengths = checkIndexAndDataFile(indexFile, dataFile, lengths.length) | ||
|
|
||
| if (existedLengths != null) { | ||
| // Use the lengths of existed output for MapStatus | ||
|
There was a problem hiding this comment. |
||
| System.arraycopy(existedLengths, 0, lengths, 0, lengths.length) | ||
| dataTmp.delete() | ||
|
There was a problem hiding this comment. though unlikely, will this throw NPE? |
||
| indexTmp.delete() | ||
| } else { | ||
| if (indexFile.exists()) { | ||
|
There was a problem hiding this comment. |
||
| indexFile.delete() | ||
| } | ||
| if (dataFile.exists()) { | ||
| dataFile.delete() | ||
| } | ||
| if (!indexTmp.renameTo(indexFile)) { | ||
| throw new IOException("fail to rename data file " + indexTmp + " to " + indexFile) | ||
|
|
||
| } | ||
| if (dataTmp != null && dataTmp.exists() && !dataTmp.renameTo(dataFile)) { | ||
| throw new IOException("fail to rename data file " + dataTmp + " to " + dataFile) | ||
|
There was a problem hiding this comment. I don't think there is a particular flaw here, but its a bit hard to follow since its a mix of first-attempt-wins and last-attempt wins. First attempt if there is a data file & index file; last attempt if its only an index file. the problem w/ last-attempt is that this delete will fail on windows if the file is open for reading, I believe. Though we can't get around that because SPARK-4085 always requires us to delete some files that might be open, in which case we hope that we don't run into this race again on the next retry. It would be nice to minimize that case, though. We'd be closer to first-attempt-wins if we always wrote a dataFile, even if its empty when dataTmp == null. There is also an issue w/ mapStatus & non-deterministic data. It might not matter which output you get, but the mapstatus should be consistent with the data that is read. If attempt 1 writes non-empty outputs a,b,c, and attempt 2 writes non-empty outputs d,e,f (which are not committed), the reduce tasks might get the mapstatus for attempt 2, look for outputs d,e,f, and get nothing but empty blocks. Matei had suggested writing the mapstatus to a file, so that subsequent attempts always return the map status corresponding to the first successful attempt. |
||
| } | ||
|
There was a problem hiding this comment. should we check what happens if this fails? Should we throw an exception? There was a problem hiding this comment. This will slowdown the normal path, I think it's not needed. |
||
| } | ||
| } | ||
| } | ||
|
|
||
| override def getBlockData(blockId: ShuffleBlockId): ManagedBuffer = { | ||
|
|
||
There was a problem hiding this comment.
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