SPARK-2045 Sort-based shuffle #1499
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.shuffle.sort | ||
|
|
||
| import java.io.{DataInputStream, FileInputStream} | ||
|
|
||
| import org.apache.spark.shuffle._ | ||
| import org.apache.spark.{TaskContext, ShuffleDependency} | ||
| import org.apache.spark.shuffle.hash.HashShuffleReader | ||
| import org.apache.spark.storage.{DiskBlockManager, FileSegment, ShuffleBlockId} | ||
|
|
||
| private[spark] class SortShuffleManager extends ShuffleManager { | ||
| /** | ||
| * Register a shuffle with the manager and obtain a handle for it to pass to tasks. | ||
| */ | ||
| override def registerShuffle[K, V, C]( | ||
| shuffleId: Int, | ||
| numMaps: Int, | ||
| dependency: ShuffleDependency[K, V, C]): ShuffleHandle = { | ||
| new BaseShuffleHandle(shuffleId, numMaps, dependency) | ||
| } | ||
|
|
||
| /** | ||
| * Get a reader for a range of reduce partitions (startPartition to endPartition-1, inclusive). | ||
| * Called on executors by reduce tasks. | ||
| */ | ||
| override def getReader[K, C]( | ||
| handle: ShuffleHandle, | ||
| startPartition: Int, | ||
| endPartition: Int, | ||
| context: TaskContext): ShuffleReader[K, C] = { | ||
| // We currently use the same block store shuffle fetcher as the hash-based shuffle. | ||
| new HashShuffleReader( | ||
| handle.asInstanceOf[BaseShuffleHandle[K, _, C]], startPartition, endPartition, context) | ||
| } | ||
|
|
||
| /** Get a writer for a given partition. Called on executors by map tasks. */ | ||
| override def getWriter[K, V](handle: ShuffleHandle, mapId: Int, context: TaskContext) | ||
| : ShuffleWriter[K, V] = { | ||
| new SortShuffleWriter(handle.asInstanceOf[BaseShuffleHandle[K, V, _]], mapId, context) | ||
| } | ||
|
|
||
| /** Remove a shuffle's metadata from the ShuffleManager. */ | ||
| override def unregisterShuffle(shuffleId: Int): Unit = {} | ||
|
There was a problem hiding this comment. shuffle output file in sortShuffleWritter do not get cleaned. We might need to add map to save registered shuffle handle, Then try to remove the data file in unregisterShuffle method. Though at present, in HashShuffleManager, this is also not implemented. But HashShuffleManager depends on shuffleBlockManager and the file will be cleaned there. I have a PR to generalize shuffleBlockManager and hide it behind shuffleMananger. as in #1241 , and upon blockMananger do remove shuffle, will call into this unregisterShuffle method. Will rebase upon this PR been merged. Should we fix this issue in my PR, or add the store/clean shuffleHandle logic here in this PR firstly? There was a problem hiding this comment. Good question, but files on disk actually get cleaned by BlockManager directly if cleaning is turned on, and sort-based shuffle is set up to only use files. ShuffleBlockManager had extra state in memory in the form of metadata, that's why it needs its own cleaner. But I don't think there's an issue here. It will be important to do something for unregisterShuffle, but right now nothing calls that. That's part of the API because I wanted to move the MapOutputTracker behind the ShuffleManager as well. But any patches to move more stuff behind it are welcome, and I agree the ShuffleBlockManager should be specific to hash-based shuffle. There was a problem hiding this comment. Maybe I miss some code? But from what I understanding, when the cleaning is turned on, it seems to me, if by timestamp approaching, blockManager won't remove shuffle data,ShuffleBlockManager will do the work by itself. And if by auto clean approaching when doCleanShuffle is called, it go through the current ShuffleBlockManager interface. In neither case, these sortShuffleWritter generated file will be cleaned? Or do you mean, as long as there are no metadata in memory, shuffle file on disk is ok to not be removed until application exit? There was a problem hiding this comment. Ah, I see, these files are never registered with the BlockManager so they won't be cleaned by it. I'll modify ShuffleBlockManager to clean them for now. There was a problem hiding this comment. Alright, fixed this in the latest commit. |
||
|
|
||
| /** Shut down this ShuffleManager. */ | ||
| override def stop(): Unit = {} | ||
|
|
||
| /** Get the location of a block in a map output file. Uses the index file we create for it. */ | ||
| def getBlockLocation(blockId: ShuffleBlockId, diskManager: DiskBlockManager): FileSegment = { | ||
| // The block is actually going to be a range of a single map output file for this map, so | ||
| // figure out the ID of the consolidated file, then the offset within that from our index | ||
| val consolidatedId = blockId.copy(reduceId = 0) | ||
| val indexFile = diskManager.getFile(consolidatedId.name + ".index") | ||
| val in = new DataInputStream(new FileInputStream(indexFile)) | ||
| try { | ||
| in.skip(blockId.reduceId * 8) | ||
| val offset = in.readLong() | ||
| val nextOffset = in.readLong() | ||
| new FileSegment(diskManager.getFile(consolidatedId), offset, nextOffset - offset) | ||
| } finally { | ||
| in.close() | ||
| } | ||
| } | ||
| } | ||
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.
Did you mean to add this annotation here?
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.
Yup, see the comments on mateiz@bf71388.