You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pw...@apache.org on 2014/01/14 07:56:37 UTC
[1/4] git commit: Improvements to external sorting
Updated Branches:
refs/heads/master 68641bce6 -> 945fe7a37
Improvements to external sorting
1. Adds the option of compressing outputs.
2. Adds batching to the serialization to prevent OOM on the read side.
3. Slight renaming of config options.
4. Use Spark's buffer size for reads in addition to writes.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/5d61e051
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/5d61e051
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/5d61e051
Branch: refs/heads/master
Commit: 5d61e051c2ad5955f0101de6f0ecdf5d243e4f5e
Parents: e6ed13f
Author: Patrick Wendell <pw...@gmail.com>
Authored: Mon Jan 13 11:30:09 2014 -0800
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Mon Jan 13 12:21:39 2014 -0800
----------------------------------------------------------------------
.../scala/org/apache/spark/Aggregator.scala | 2 +-
.../org/apache/spark/storage/BlockManager.scala | 3 ++
.../util/collection/ExternalAppendOnlyMap.scala | 48 ++++++++++++++++----
docs/configuration.md | 11 ++++-
4 files changed, 52 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/5d61e051/core/src/main/scala/org/apache/spark/Aggregator.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/Aggregator.scala b/core/src/main/scala/org/apache/spark/Aggregator.scala
index 8b30cd4..ee44670 100644
--- a/core/src/main/scala/org/apache/spark/Aggregator.scala
+++ b/core/src/main/scala/org/apache/spark/Aggregator.scala
@@ -32,7 +32,7 @@ case class Aggregator[K, V, C] (
mergeCombiners: (C, C) => C) {
private val sparkConf = SparkEnv.get.conf
- private val externalSorting = sparkConf.getBoolean("spark.shuffle.externalSorting", true)
+ private val externalSorting = sparkConf.getBoolean("spark.shuffle.external", true)
def combineValuesByKey(iter: Iterator[_ <: Product2[K, V]]) : Iterator[(K, C)] = {
if (!externalSorting) {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/5d61e051/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 6f1345c..0e770ed 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -80,6 +80,8 @@ private[spark] class BlockManager(
val compressShuffle = conf.getBoolean("spark.shuffle.compress", true)
// Whether to compress RDD partitions that are stored serialized
val compressRdds = conf.getBoolean("spark.rdd.compress", false)
+ // Whether to compress shuffle output temporarily spilled to disk
+ val compressExternalShuffle = conf.getBoolean("spark.shuffle.external.compress", false)
val heartBeatFrequency = BlockManager.getHeartBeatFrequency(conf)
@@ -790,6 +792,7 @@ private[spark] class BlockManager(
case ShuffleBlockId(_, _, _) => compressShuffle
case BroadcastBlockId(_) => compressBroadcast
case RDDBlockId(_, _) => compressRdds
+ case TempBlockId(_) => compressExternalShuffle
case _ => false
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/5d61e051/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala b/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
index e3bcd89..fd17413 100644
--- a/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
+++ b/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
@@ -27,7 +27,7 @@ import scala.collection.mutable.ArrayBuffer
import org.apache.spark.{Logging, SparkEnv}
import org.apache.spark.serializer.Serializer
-import org.apache.spark.storage.{DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
/**
* An append-only map that spills sorted content to disk when there is insufficient space for it
@@ -60,7 +60,7 @@ private[spark] class ExternalAppendOnlyMap[K, V, C](
mergeValue: (C, V) => C,
mergeCombiners: (C, C) => C,
serializer: Serializer = SparkEnv.get.serializerManager.default,
- diskBlockManager: DiskBlockManager = SparkEnv.get.blockManager.diskBlockManager)
+ blockManager: BlockManager = SparkEnv.get.blockManager)
extends Iterable[(K, C)] with Serializable with Logging {
import ExternalAppendOnlyMap._
@@ -68,6 +68,7 @@ private[spark] class ExternalAppendOnlyMap[K, V, C](
private var currentMap = new SizeTrackingAppendOnlyMap[K, C]
private val spilledMaps = new ArrayBuffer[DiskMapIterator]
private val sparkConf = SparkEnv.get.conf
+ private val diskBlockManager = blockManager.diskBlockManager
// Collective memory threshold shared across all running tasks
private val maxMemoryThreshold = {
@@ -82,6 +83,14 @@ private[spark] class ExternalAppendOnlyMap[K, V, C](
// Number of in-memory pairs inserted before tracking the map's shuffle memory usage
private val trackMemoryThreshold = 1000
+ // Size of object batches when reading/writing from serializers. Objects are written in
+ // batches, with each batch using its own serialization stream. This cuts down on the size
+ // of reference-tracking maps constructed when deserializing a stream.
+ //
+ // NOTE: Setting this too low can cause excess copying when serializing, since some serailizers
+ // grow internal data structures by growing + copying every time the number of objects doubles.
+ private val serializerBatchSize = sparkConf.getLong("spark.shuffle.external.batchSize", 10000)
+
// How many times we have spilled so far
private var spillCount = 0
@@ -139,21 +148,34 @@ private[spark] class ExternalAppendOnlyMap[K, V, C](
logWarning("Spilling in-memory map of %d MB to disk (%d time%s so far)"
.format(mapSize / (1024 * 1024), spillCount, if (spillCount > 1) "s" else ""))
val (blockId, file) = diskBlockManager.createTempBlock()
- val writer =
- new DiskBlockObjectWriter(blockId, file, serializer, fileBufferSize, identity, syncWrites)
+
+ val compressStream: OutputStream => OutputStream = blockManager.wrapForCompression(blockId, _)
+ def getNewWriter = new DiskBlockObjectWriter(blockId, file, serializer, fileBufferSize,
+ compressStream, syncWrites)
+
+ var writer = getNewWriter
+ var objectsWritten = 0
try {
val it = currentMap.destructiveSortedIterator(comparator)
while (it.hasNext) {
val kv = it.next()
writer.write(kv)
+ objectsWritten += 1
+
+ if (objectsWritten == serializerBatchSize) {
+ writer.commit()
+ writer = getNewWriter
+ objectsWritten = 0
+ }
}
- writer.commit()
+
+ if (objectsWritten > 0) writer.commit()
} finally {
// Partial failures cannot be tolerated; do not revert partial writes
writer.close()
}
currentMap = new SizeTrackingAppendOnlyMap[K, C]
- spilledMaps.append(new DiskMapIterator(file))
+ spilledMaps.append(new DiskMapIterator(file, blockId))
// Reset the amount of shuffle memory used by this map in the global pool
val shuffleMemoryMap = SparkEnv.get.shuffleMemoryMap
@@ -297,16 +319,24 @@ private[spark] class ExternalAppendOnlyMap[K, V, C](
/**
* An iterator that returns (K, C) pairs in sorted order from an on-disk map
*/
- private class DiskMapIterator(file: File) extends Iterator[(K, C)] {
+ private class DiskMapIterator(file: File, blockId: BlockId) extends Iterator[(K, C)] {
val fileStream = new FileInputStream(file)
- val bufferedStream = new FastBufferedInputStream(fileStream)
- val deserializeStream = ser.deserializeStream(bufferedStream)
+ val bufferedStream = new FastBufferedInputStream(fileStream, fileBufferSize)
+ val compressedStream = blockManager.wrapForCompression(blockId, bufferedStream)
+ var deserializeStream = ser.deserializeStream(compressedStream)
+ var objectsRead = 0
+
var nextItem: (K, C) = null
var eof = false
def readNextItem(): (K, C) = {
if (!eof) {
try {
+ if (objectsRead == serializerBatchSize) {
+ deserializeStream = ser.deserializeStream(compressedStream)
+ objectsRead = 0
+ }
+ objectsRead += 1
return deserializeStream.readObject().asInstanceOf[(K, C)]
} catch {
case e: EOFException =>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/5d61e051/docs/configuration.md
----------------------------------------------------------------------
diff --git a/docs/configuration.md b/docs/configuration.md
index 40a57c4..350e314 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -116,7 +116,7 @@ Apart from these, the following properties are also available, and may be useful
<td>0.3</td>
<td>
Fraction of Java heap to use for aggregation and cogroups during shuffles, if
- <code>spark.shuffle.externalSorting</code> is enabled. At any given time, the collective size of
+ <code>spark.shuffle.external</code> is true. At any given time, the collective size of
all in-memory maps used for shuffles is bounded by this limit, beyond which the contents will
begin to spill to disk. If spills are often, consider increasing this value at the expense of
<code>spark.storage.memoryFraction</code>.
@@ -155,6 +155,13 @@ Apart from these, the following properties are also available, and may be useful
</td>
</tr>
<tr>
+ <td>spark.shuffle.external.compress</td>
+ <td>false</td>
+ <td>
+ Whether to compress data spilled during shuffles.
+ </td>
+</tr>
+<tr>
<td>spark.broadcast.compress</td>
<td>true</td>
<td>
@@ -388,7 +395,7 @@ Apart from these, the following properties are also available, and may be useful
</td>
</tr>
<tr>
- <td>spark.shuffle.externalSorting</td>
+ <td>spark.shuffle.external</td>
<td>true</td>
<td>
If set to "true", limits the amount of memory used during reduces by spilling data out to disk. This spilling
[3/4] git commit: Fix for Kryo Serializer
Posted by pw...@apache.org.
Fix for Kryo Serializer
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/d4cd5deb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/d4cd5deb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/d4cd5deb
Branch: refs/heads/master
Commit: d4cd5debf4ac11884e4e703f482ffc17427e277f
Parents: c3816de
Author: Patrick Wendell <pw...@gmail.com>
Authored: Mon Jan 13 19:03:59 2014 -0800
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Mon Jan 13 19:03:59 2014 -0800
----------------------------------------------------------------------
.../util/collection/ExternalAppendOnlyMap.scala | 15 +++++++++++++--
1 file changed, 13 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/d4cd5deb/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala b/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
index 2eef6a7..2cf46e8 100644
--- a/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
+++ b/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
@@ -26,7 +26,7 @@ import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import org.apache.spark.{Logging, SparkEnv}
-import org.apache.spark.serializer.Serializer
+import org.apache.spark.serializer.{KryoDeserializationStream, KryoSerializationStream, Serializer}
import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
/**
@@ -333,7 +333,18 @@ private[spark] class ExternalAppendOnlyMap[K, V, C](
if (!eof) {
try {
if (objectsRead == serializerBatchSize) {
- deserializeStream = ser.deserializeStream(compressedStream)
+ val newInputStream = deserializeStream match {
+ case stream: KryoDeserializationStream =>
+ // Kryo's serializer stores an internal buffer that pre-fetches from the underlying
+ // stream. We need to capture this buffer and feed it to the new serialization
+ // stream so that the bytes are not lost.
+ val kryoInput = stream.input
+ val remainingBytes = kryoInput.limit() - kryoInput.position()
+ val extraBuf = kryoInput.readBytes(remainingBytes)
+ new SequenceInputStream(new ByteArrayInputStream(extraBuf), compressedStream)
+ case _ => compressedStream
+ }
+ deserializeStream = ser.deserializeStream(newInputStream)
objectsRead = 0
}
objectsRead += 1
[2/4] git commit: Changing option wording per discussion with Andrew
Posted by pw...@apache.org.
Changing option wording per discussion with Andrew
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/c3816de5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/c3816de5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/c3816de5
Branch: refs/heads/master
Commit: c3816de5040e3c48e58ed4762d2f4eb606812938
Parents: 5d61e05
Author: Patrick Wendell <pw...@gmail.com>
Authored: Mon Jan 13 13:24:04 2014 -0800
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Mon Jan 13 13:25:06 2014 -0800
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/Aggregator.scala | 2 +-
.../src/main/scala/org/apache/spark/storage/BlockManager.scala | 4 ++--
.../apache/spark/util/collection/ExternalAppendOnlyMap.scala | 4 ++--
docs/configuration.md | 6 +++---
4 files changed, 8 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c3816de5/core/src/main/scala/org/apache/spark/Aggregator.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/Aggregator.scala b/core/src/main/scala/org/apache/spark/Aggregator.scala
index ee44670..3807370 100644
--- a/core/src/main/scala/org/apache/spark/Aggregator.scala
+++ b/core/src/main/scala/org/apache/spark/Aggregator.scala
@@ -32,7 +32,7 @@ case class Aggregator[K, V, C] (
mergeCombiners: (C, C) => C) {
private val sparkConf = SparkEnv.get.conf
- private val externalSorting = sparkConf.getBoolean("spark.shuffle.external", true)
+ private val externalSorting = sparkConf.getBoolean("spark.shuffle.spill", true)
def combineValuesByKey(iter: Iterator[_ <: Product2[K, V]]) : Iterator[(K, C)] = {
if (!externalSorting) {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c3816de5/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 0e770ed..6461dee 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -81,7 +81,7 @@ private[spark] class BlockManager(
// Whether to compress RDD partitions that are stored serialized
val compressRdds = conf.getBoolean("spark.rdd.compress", false)
// Whether to compress shuffle output temporarily spilled to disk
- val compressExternalShuffle = conf.getBoolean("spark.shuffle.external.compress", false)
+ val compressShuffleSpill = conf.getBoolean("spark.shuffle.spill.compress", false)
val heartBeatFrequency = BlockManager.getHeartBeatFrequency(conf)
@@ -792,7 +792,7 @@ private[spark] class BlockManager(
case ShuffleBlockId(_, _, _) => compressShuffle
case BroadcastBlockId(_) => compressBroadcast
case RDDBlockId(_, _) => compressRdds
- case TempBlockId(_) => compressExternalShuffle
+ case TempBlockId(_) => compressShuffleSpill
case _ => false
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c3816de5/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala b/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
index fd17413..2eef6a7 100644
--- a/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
+++ b/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
@@ -87,9 +87,9 @@ private[spark] class ExternalAppendOnlyMap[K, V, C](
// batches, with each batch using its own serialization stream. This cuts down on the size
// of reference-tracking maps constructed when deserializing a stream.
//
- // NOTE: Setting this too low can cause excess copying when serializing, since some serailizers
+ // NOTE: Setting this too low can cause excess copying when serializing, since some serializers
// grow internal data structures by growing + copying every time the number of objects doubles.
- private val serializerBatchSize = sparkConf.getLong("spark.shuffle.external.batchSize", 10000)
+ private val serializerBatchSize = sparkConf.getLong("spark.shuffle.spill.batchSize", 10000)
// How many times we have spilled so far
private var spillCount = 0
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c3816de5/docs/configuration.md
----------------------------------------------------------------------
diff --git a/docs/configuration.md b/docs/configuration.md
index 350e314..be06bd1 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -116,7 +116,7 @@ Apart from these, the following properties are also available, and may be useful
<td>0.3</td>
<td>
Fraction of Java heap to use for aggregation and cogroups during shuffles, if
- <code>spark.shuffle.external</code> is true. At any given time, the collective size of
+ <code>spark.shuffle.spill</code> is true. At any given time, the collective size of
all in-memory maps used for shuffles is bounded by this limit, beyond which the contents will
begin to spill to disk. If spills are often, consider increasing this value at the expense of
<code>spark.storage.memoryFraction</code>.
@@ -155,7 +155,7 @@ Apart from these, the following properties are also available, and may be useful
</td>
</tr>
<tr>
- <td>spark.shuffle.external.compress</td>
+ <td>spark.shuffle.spill.compress</td>
<td>false</td>
<td>
Whether to compress data spilled during shuffles.
@@ -395,7 +395,7 @@ Apart from these, the following properties are also available, and may be useful
</td>
</tr>
<tr>
- <td>spark.shuffle.external</td>
+ <td>spark.shuffle.spill</td>
<td>true</td>
<td>
If set to "true", limits the amount of memory used during reduces by spilling data out to disk. This spilling
[4/4] git commit: Merge pull request #408 from
pwendell/external-serializers
Posted by pw...@apache.org.
Merge pull request #408 from pwendell/external-serializers
Improvements to external sorting
1. Adds the option of compressing outputs.
2. Adds batching to the serialization to prevent OOM on the read side.
3. Slight renaming of config options.
4. Use Spark's buffer size for reads in addition to writes.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/945fe7a3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/945fe7a3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/945fe7a3
Branch: refs/heads/master
Commit: 945fe7a37ea3189b5a9f8a74e5c2fa9c1088ebfc
Parents: 68641bc d4cd5de
Author: Patrick Wendell <pw...@gmail.com>
Authored: Mon Jan 13 22:56:12 2014 -0800
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Mon Jan 13 22:56:12 2014 -0800
----------------------------------------------------------------------
.../scala/org/apache/spark/Aggregator.scala | 2 +-
.../org/apache/spark/storage/BlockManager.scala | 3 +
.../util/collection/ExternalAppendOnlyMap.scala | 61 ++++++++++++++++----
docs/configuration.md | 11 +++-
4 files changed, 64 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/945fe7a3/core/src/main/scala/org/apache/spark/Aggregator.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/org/apache/spark/Aggregator.scala
index cda3a95,3807370..6d439fd
--- a/core/src/main/scala/org/apache/spark/Aggregator.scala
+++ b/core/src/main/scala/org/apache/spark/Aggregator.scala
@@@ -32,10 -32,9 +32,10 @@@ case class Aggregator[K, V, C]
mergeCombiners: (C, C) => C) {
private val sparkConf = SparkEnv.get.conf
- private val externalSorting = sparkConf.getBoolean("spark.shuffle.externalSorting", true)
+ private val externalSorting = sparkConf.getBoolean("spark.shuffle.spill", true)
- def combineValuesByKey(iter: Iterator[_ <: Product2[K, V]]) : Iterator[(K, C)] = {
+ def combineValuesByKey(iter: Iterator[_ <: Product2[K, V]],
+ context: TaskContext) : Iterator[(K, C)] = {
if (!externalSorting) {
val combiners = new AppendOnlyMap[K,C]
var kv: Product2[K, V] = null
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/945fe7a3/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
index c63f47c,2cf46e8..64e9b43
--- a/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
+++ b/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
@@@ -150,11 -160,18 +164,19 @@@ private[spark] class ExternalAppendOnly
while (it.hasNext) {
val kv = it.next()
writer.write(kv)
+ objectsWritten += 1
+
+ if (objectsWritten == serializerBatchSize) {
+ writer.commit()
+ writer = getNewWriter
+ objectsWritten = 0
+ }
}
- writer.commit()
+
+ if (objectsWritten > 0) writer.commit()
} finally {
// Partial failures cannot be tolerated; do not revert partial writes
+ _diskBytesSpilled += writer.bytesWritten
writer.close()
}
currentMap = new SizeTrackingAppendOnlyMap[K, C]