You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by rx...@apache.org on 2013/10/28 03:11:26 UTC
[1/3] git commit: Eliminate extra memory usage when shuffle file
consolidation is disabled
Updated Branches:
refs/heads/master e018f2d0a -> d4df4749a
Eliminate extra memory usage when shuffle file consolidation is disabled
Otherwise, we see SPARK-946 even when shuffle file consolidation is disabled.
Fixing SPARK-946 is still forthcoming.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/596f1847
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/596f1847
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/596f1847
Branch: refs/heads/master
Commit: 596f18479e5cb14fdc0af800f97b7b13e963fe5b
Parents: e018f2d
Author: Aaron Davidson <aa...@databricks.com>
Authored: Sat Oct 26 22:25:24 2013 -0700
Committer: Aaron Davidson <aa...@databricks.com>
Committed: Sat Oct 26 22:35:01 2013 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/storage/BlockManager.scala | 5 ++++-
.../org/apache/spark/storage/ShuffleBlockManager.scala | 13 +++++++++----
2 files changed, 13 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/596f1847/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
index e6329cb..c02c032 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -523,7 +523,10 @@ private[spark] class BlockManager(
val file = diskBlockManager.createBlockFile(blockId, filename, allowAppending = true)
val writer = new DiskBlockObjectWriter(blockId, file, serializer, bufferSize, compressStream)
writer.registerCloseEventHandler(() => {
- diskBlockManager.mapBlockToFileSegment(blockId, writer.fileSegment())
+ if (blockId.name != filename) {
+ // Avoid producing mappings if we're using unique filenames for blocks.
+ diskBlockManager.mapBlockToFileSegment(blockId, writer.fileSegment())
+ }
val myInfo = new BlockInfo(StorageLevel.DISK_ONLY, false)
blockInfo.put(blockId, myInfo)
myInfo.markReady(writer.fileSegment().length)
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/596f1847/core/src/main/scala/org/apache/spark/storage/ShuffleBlockManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/ShuffleBlockManager.scala b/core/src/main/scala/org/apache/spark/storage/ShuffleBlockManager.scala
index 229178c..066e45a 100644
--- a/core/src/main/scala/org/apache/spark/storage/ShuffleBlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/ShuffleBlockManager.scala
@@ -63,8 +63,12 @@ class ShuffleBlockManager(blockManager: BlockManager) {
val fileId = getUnusedFileId()
val writers = Array.tabulate[BlockObjectWriter](numBuckets) { bucketId =>
val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
- val filename = physicalFileName(shuffleId, bucketId, fileId)
- blockManager.getDiskWriter(blockId, filename, serializer, bufferSize)
+ if (consolidateShuffleFiles) {
+ val filename = physicalFileName(shuffleId, bucketId, fileId)
+ blockManager.getDiskWriter(blockId, filename, serializer, bufferSize)
+ } else {
+ blockManager.getDiskWriter(blockId, blockId.name, serializer, bufferSize)
+ }
}
new ShuffleWriterGroup(mapId, fileId, writers)
}
@@ -81,8 +85,9 @@ class ShuffleBlockManager(blockManager: BlockManager) {
}
private def recycleFileId(fileId: Int) {
- if (!consolidateShuffleFiles) { return } // ensures we always generate new file id
- unusedFileIds.add(fileId)
+ if (consolidateShuffleFiles) {
+ unusedFileIds.add(fileId)
+ }
}
private def physicalFileName(shuffleId: Int, bucketId: Int, fileId: Int) = {
[3/3] git commit: Merge pull request #115 from aarondav/shuffle-fix
Posted by rx...@apache.org.
Merge pull request #115 from aarondav/shuffle-fix
Eliminate extra memory usage when shuffle file consolidation is disabled
Otherwise, we see SPARK-946 even when shuffle file consolidation is disabled.
Fixing SPARK-946 is still forthcoming.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/d4df4749
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/d4df4749
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/d4df4749
Branch: refs/heads/master
Commit: d4df4749a8738628aa79802c4fb2805d02f47bb8
Parents: e018f2d 4261e83
Author: Reynold Xin <rx...@apache.org>
Authored: Sun Oct 27 22:11:21 2013 -0400
Committer: Reynold Xin <rx...@apache.org>
Committed: Sun Oct 27 22:11:21 2013 -0400
----------------------------------------------------------------------
.../scala/org/apache/spark/storage/BlockManager.scala | 4 +++-
.../org/apache/spark/storage/ShuffleBlockManager.scala | 13 +++++++++----
2 files changed, 12 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
[2/3] git commit: Use flag instead of name check.
Posted by rx...@apache.org.
Use flag instead of name check.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/4261e834
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/4261e834
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/4261e834
Branch: refs/heads/master
Commit: 4261e834cb9c446366569a8dd49d3ba7c1e72ffc
Parents: 596f184
Author: Aaron Davidson <aa...@databricks.com>
Authored: Sat Oct 26 23:53:38 2013 -0700
Committer: Aaron Davidson <aa...@databricks.com>
Committed: Sat Oct 26 23:53:38 2013 -0700
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/storage/BlockManager.scala | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/4261e834/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
index c02c032..dbe573d 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -523,8 +523,7 @@ private[spark] class BlockManager(
val file = diskBlockManager.createBlockFile(blockId, filename, allowAppending = true)
val writer = new DiskBlockObjectWriter(blockId, file, serializer, bufferSize, compressStream)
writer.registerCloseEventHandler(() => {
- if (blockId.name != filename) {
- // Avoid producing mappings if we're using unique filenames for blocks.
+ if (shuffleBlockManager.consolidateShuffleFiles) {
diskBlockManager.mapBlockToFileSegment(blockId, writer.fileSegment())
}
val myInfo = new BlockInfo(StorageLevel.DISK_ONLY, false)