diff options
10 files changed, 59 insertions, 59 deletions
diff --git a/core/src/main/scala/spark/KryoSerializer.scala b/core/src/main/scala/spark/KryoSerializer.scala index 4b95e05bd3..244d50f49c 100644 --- a/core/src/main/scala/spark/KryoSerializer.scala +++ b/core/src/main/scala/spark/KryoSerializer.scala @@ -192,8 +192,8 @@ class KryoSerializer extends Serializer with Logging { (1, 1, 1), (1, 1, 1, 1), (1, 1, 1, 1, 1), None, ByteBuffer.allocate(1), - StorageLevel.MEMORY_ONLY_DESER, - PutBlock("1", ByteBuffer.allocate(1), StorageLevel.MEMORY_ONLY_DESER), + StorageLevel.MEMORY_ONLY, + PutBlock("1", ByteBuffer.allocate(1), StorageLevel.MEMORY_ONLY), GotBlock("1", ByteBuffer.allocate(1)), GetBlock("1") ) diff --git a/core/src/main/scala/spark/RDD.scala b/core/src/main/scala/spark/RDD.scala index cce0ea2183..b0d86ebbae 100644 --- a/core/src/main/scala/spark/RDD.scala +++ b/core/src/main/scala/spark/RDD.scala @@ -89,14 +89,14 @@ abstract class RDD[T: ClassManifest](@transient sc: SparkContext) extends Serial } // Turn on the default caching level for this RDD - def persist(): RDD[T] = persist(StorageLevel.MEMORY_ONLY_DESER) + def persist(): RDD[T] = persist(StorageLevel.MEMORY_ONLY) // Turn on the default caching level for this RDD def cache(): RDD[T] = persist() def getStorageLevel = storageLevel - def checkpoint(level: StorageLevel = StorageLevel.DISK_AND_MEMORY_DESER_2): RDD[T] = { + def checkpoint(level: StorageLevel = StorageLevel.MEMORY_AND_DISK_2): RDD[T] = { if (!level.useDisk && level.replication < 2) { throw new Exception("Cannot checkpoint without using disk or replication (level requested was " + level + ")") } diff --git a/core/src/main/scala/spark/broadcast/BitTorrentBroadcast.scala b/core/src/main/scala/spark/broadcast/BitTorrentBroadcast.scala index 016dc00fb0..0bb9937992 100644 --- a/core/src/main/scala/spark/broadcast/BitTorrentBroadcast.scala +++ b/core/src/main/scala/spark/broadcast/BitTorrentBroadcast.scala @@ -18,7 +18,7 @@ extends Broadcast[T] with Logging with Serializable { MultiTracker.synchronized { SparkEnv.get.blockManager.putSingle( - uuid.toString, value_, StorageLevel.MEMORY_ONLY_DESER, false) + uuid.toString, value_, StorageLevel.MEMORY_ONLY, false) } @transient var arrayOfBlocks: Array[BroadcastBlock] = null @@ -135,7 +135,7 @@ extends Broadcast[T] with Logging with Serializable { if (receptionSucceeded) { value_ = MultiTracker.unBlockifyObject[T](arrayOfBlocks, totalBytes, totalBlocks) SparkEnv.get.blockManager.putSingle( - uuid.toString, value_, StorageLevel.MEMORY_ONLY_DESER, false) + uuid.toString, value_, StorageLevel.MEMORY_ONLY, false) } else { logError("Reading Broadcasted variable " + uuid + " failed") } diff --git a/core/src/main/scala/spark/broadcast/HttpBroadcast.scala b/core/src/main/scala/spark/broadcast/HttpBroadcast.scala index eacf237508..a98cbb6994 100644 --- a/core/src/main/scala/spark/broadcast/HttpBroadcast.scala +++ b/core/src/main/scala/spark/broadcast/HttpBroadcast.scala @@ -19,7 +19,7 @@ extends Broadcast[T] with Logging with Serializable { HttpBroadcast.synchronized { SparkEnv.get.blockManager.putSingle( - uuid.toString, value_, StorageLevel.MEMORY_ONLY_DESER, false) + uuid.toString, value_, StorageLevel.MEMORY_ONLY, false) } if (!isLocal) { @@ -37,7 +37,7 @@ extends Broadcast[T] with Logging with Serializable { val start = System.nanoTime value_ = HttpBroadcast.read[T](uuid) SparkEnv.get.blockManager.putSingle( - uuid.toString, value_, StorageLevel.MEMORY_ONLY_DESER, false) + uuid.toString, value_, StorageLevel.MEMORY_ONLY, false) val time = (System.nanoTime - start) / 1e9 logInfo("Reading broadcast variable " + uuid + " took " + time + " s") } diff --git a/core/src/main/scala/spark/broadcast/TreeBroadcast.scala b/core/src/main/scala/spark/broadcast/TreeBroadcast.scala index c9e1e67d87..3b54d570be 100644 --- a/core/src/main/scala/spark/broadcast/TreeBroadcast.scala +++ b/core/src/main/scala/spark/broadcast/TreeBroadcast.scala @@ -17,7 +17,7 @@ extends Broadcast[T] with Logging with Serializable { MultiTracker.synchronized { SparkEnv.get.blockManager.putSingle( - uuid.toString, value_, StorageLevel.MEMORY_ONLY_DESER, false) + uuid.toString, value_, StorageLevel.MEMORY_ONLY, false) } @transient var arrayOfBlocks: Array[BroadcastBlock] = null @@ -112,7 +112,7 @@ extends Broadcast[T] with Logging with Serializable { if (receptionSucceeded) { value_ = MultiTracker.unBlockifyObject[T](arrayOfBlocks, totalBytes, totalBlocks) SparkEnv.get.blockManager.putSingle( - uuid.toString, value_, StorageLevel.MEMORY_ONLY_DESER, false) + uuid.toString, value_, StorageLevel.MEMORY_ONLY, false) } else { logError("Reading Broadcasted variable " + uuid + " failed") } diff --git a/core/src/main/scala/spark/storage/BlockMessage.scala b/core/src/main/scala/spark/storage/BlockMessage.scala index 5e2ccb199a..4b5cfebba2 100644 --- a/core/src/main/scala/spark/storage/BlockMessage.scala +++ b/core/src/main/scala/spark/storage/BlockMessage.scala @@ -196,7 +196,7 @@ object BlockMessage { def main(args: Array[String]) { val B = new BlockMessage() - B.set(new PutBlock("ABC", ByteBuffer.allocate(10), StorageLevel.DISK_AND_MEMORY_2)) + B.set(new PutBlock("ABC", ByteBuffer.allocate(10), StorageLevel.MEMORY_AND_DISK_SER_2)) val bMsg = B.toBufferMessage val C = new BlockMessage() C.set(bMsg) diff --git a/core/src/main/scala/spark/storage/BlockMessageArray.scala b/core/src/main/scala/spark/storage/BlockMessageArray.scala index 928857056f..64acc7eb47 100644 --- a/core/src/main/scala/spark/storage/BlockMessageArray.scala +++ b/core/src/main/scala/spark/storage/BlockMessageArray.scala @@ -99,7 +99,7 @@ object BlockMessageArray { if (i % 2 == 0) { val buffer = ByteBuffer.allocate(100) buffer.clear - BlockMessage.fromPutBlock(PutBlock(i.toString, buffer, StorageLevel.MEMORY_ONLY)) + BlockMessage.fromPutBlock(PutBlock(i.toString, buffer, StorageLevel.MEMORY_ONLY_SER)) } else { BlockMessage.fromGetBlock(GetBlock(i.toString)) } diff --git a/core/src/main/scala/spark/storage/StorageLevel.scala b/core/src/main/scala/spark/storage/StorageLevel.scala index b168c8e869..2d52fac1ef 100644 --- a/core/src/main/scala/spark/storage/StorageLevel.scala +++ b/core/src/main/scala/spark/storage/StorageLevel.scala @@ -67,12 +67,12 @@ object StorageLevel { val NONE = new StorageLevel(false, false, false) val DISK_ONLY = new StorageLevel(true, false, false) val DISK_ONLY_2 = new StorageLevel(true, false, false, 2) - val MEMORY_ONLY = new StorageLevel(false, true, false) - val MEMORY_ONLY_2 = new StorageLevel(false, true, false, 2) - val MEMORY_ONLY_DESER = new StorageLevel(false, true, true) - val MEMORY_ONLY_DESER_2 = new StorageLevel(false, true, true, 2) - val DISK_AND_MEMORY = new StorageLevel(true, true, false) - val DISK_AND_MEMORY_2 = new StorageLevel(true, true, false, 2) - val DISK_AND_MEMORY_DESER = new StorageLevel(true, true, true) - val DISK_AND_MEMORY_DESER_2 = new StorageLevel(true, true, true, 2) + val MEMORY_ONLY = new StorageLevel(false, true, true) + val MEMORY_ONLY_2 = new StorageLevel(false, true, true, 2) + val MEMORY_ONLY_SER = new StorageLevel(false, true, false) + val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, 2) + val MEMORY_AND_DISK = new StorageLevel(true, true, true) + val MEMORY_AND_DISK_2 = new StorageLevel(true, true, true, 2) + val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false) + val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, 2) } diff --git a/core/src/test/scala/spark/storage/BlockManagerSuite.scala b/core/src/test/scala/spark/storage/BlockManagerSuite.scala index 9e55647bd0..c5e0f7be06 100644 --- a/core/src/test/scala/spark/storage/BlockManagerSuite.scala +++ b/core/src/test/scala/spark/storage/BlockManagerSuite.scala @@ -55,9 +55,9 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT val a3 = new Array[Byte](400) // Putting a1, a2 and a3 in memory and telling master only about a1 and a2 - store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY_DESER) - store.putSingle("a2", a2, StorageLevel.MEMORY_ONLY_DESER) - store.putSingle("a3", a3, StorageLevel.MEMORY_ONLY_DESER, false) + store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY) + store.putSingle("a2", a2, StorageLevel.MEMORY_ONLY) + store.putSingle("a3", a3, StorageLevel.MEMORY_ONLY, false) // Checking whether blocks are in memory assert(store.getSingle("a1") != None, "a1 was not in store") @@ -83,16 +83,16 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT val a1 = new Array[Byte](400) val a2 = new Array[Byte](400) val a3 = new Array[Byte](400) - store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY_DESER) - store.putSingle("a2", a2, StorageLevel.MEMORY_ONLY_DESER) - store.putSingle("a3", a3, StorageLevel.MEMORY_ONLY_DESER) + store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY) + store.putSingle("a2", a2, StorageLevel.MEMORY_ONLY) + store.putSingle("a3", a3, StorageLevel.MEMORY_ONLY) assert(store.getSingle("a2") != None, "a2 was not in store") assert(store.getSingle("a3") != None, "a3 was not in store") Thread.sleep(100) assert(store.getSingle("a1") === None, "a1 was in store") assert(store.getSingle("a2") != None, "a2 was not in store") // At this point a2 was gotten last, so LRU will getSingle rid of a3 - store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY_DESER) + store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY) assert(store.getSingle("a1") != None, "a1 was not in store") assert(store.getSingle("a2") != None, "a2 was not in store") Thread.sleep(100) @@ -104,16 +104,16 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT val a1 = new Array[Byte](400) val a2 = new Array[Byte](400) val a3 = new Array[Byte](400) - store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY) - store.putSingle("a2", a2, StorageLevel.MEMORY_ONLY) - store.putSingle("a3", a3, StorageLevel.MEMORY_ONLY) + store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY_SER) + store.putSingle("a2", a2, StorageLevel.MEMORY_ONLY_SER) + store.putSingle("a3", a3, StorageLevel.MEMORY_ONLY_SER) Thread.sleep(100) assert(store.getSingle("a2") != None, "a2 was not in store") assert(store.getSingle("a3") != None, "a3 was not in store") assert(store.getSingle("a1") === None, "a1 was in store") assert(store.getSingle("a2") != None, "a2 was not in store") // At this point a2 was gotten last, so LRU will getSingle rid of a3 - store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY_DESER) + store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY) Thread.sleep(100) assert(store.getSingle("a1") != None, "a1 was not in store") assert(store.getSingle("a2") != None, "a2 was not in store") @@ -138,9 +138,9 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT val a1 = new Array[Byte](400) val a2 = new Array[Byte](400) val a3 = new Array[Byte](400) - store.putSingle("a1", a1, StorageLevel.DISK_AND_MEMORY_DESER) - store.putSingle("a2", a2, StorageLevel.DISK_AND_MEMORY_DESER) - store.putSingle("a3", a3, StorageLevel.DISK_AND_MEMORY_DESER) + store.putSingle("a1", a1, StorageLevel.MEMORY_AND_DISK) + store.putSingle("a2", a2, StorageLevel.MEMORY_AND_DISK) + store.putSingle("a3", a3, StorageLevel.MEMORY_AND_DISK) Thread.sleep(100) assert(store.getSingle("a2") != None, "a2 was not in store") assert(store.getSingle("a3") != None, "a3 was not in store") @@ -152,9 +152,9 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT val a1 = new Array[Byte](400) val a2 = new Array[Byte](400) val a3 = new Array[Byte](400) - store.putSingle("a1", a1, StorageLevel.DISK_AND_MEMORY) - store.putSingle("a2", a2, StorageLevel.DISK_AND_MEMORY) - store.putSingle("a3", a3, StorageLevel.DISK_AND_MEMORY) + store.putSingle("a1", a1, StorageLevel.MEMORY_AND_DISK_SER) + store.putSingle("a2", a2, StorageLevel.MEMORY_AND_DISK_SER) + store.putSingle("a3", a3, StorageLevel.MEMORY_AND_DISK_SER) Thread.sleep(100) assert(store.getSingle("a2") != None, "a2 was not in store") assert(store.getSingle("a3") != None, "a3 was not in store") @@ -168,8 +168,8 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT val a3 = new Array[Byte](400) val a4 = new Array[Byte](400) // First store a1 and a2, both in memory, and a3, on disk only - store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY) - store.putSingle("a2", a2, StorageLevel.MEMORY_ONLY) + store.putSingle("a1", a1, StorageLevel.MEMORY_ONLY_SER) + store.putSingle("a2", a2, StorageLevel.MEMORY_ONLY_SER) store.putSingle("a3", a3, StorageLevel.DISK_ONLY) // At this point LRU should not kick in because a3 is only on disk assert(store.getSingle("a1") != None, "a2 was not in store") @@ -179,7 +179,7 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT assert(store.getSingle("a2") != None, "a3 was not in store") assert(store.getSingle("a3") != None, "a1 was not in store") // Now let's add in a4, which uses both disk and memory; a1 should drop out - store.putSingle("a4", a4, StorageLevel.DISK_AND_MEMORY) + store.putSingle("a4", a4, StorageLevel.MEMORY_AND_DISK_SER) Thread.sleep(100) assert(store.getSingle("a1") == None, "a1 was in store") assert(store.getSingle("a2") != None, "a2 was not in store") @@ -192,9 +192,9 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT val list1 = List(new Array[Byte](200), new Array[Byte](200)) val list2 = List(new Array[Byte](200), new Array[Byte](200)) val list3 = List(new Array[Byte](200), new Array[Byte](200)) - store.put("list1", list1.iterator, StorageLevel.MEMORY_ONLY_DESER) - store.put("list2", list2.iterator, StorageLevel.MEMORY_ONLY_DESER) - store.put("list3", list3.iterator, StorageLevel.MEMORY_ONLY_DESER) + store.put("list1", list1.iterator, StorageLevel.MEMORY_ONLY) + store.put("list2", list2.iterator, StorageLevel.MEMORY_ONLY) + store.put("list3", list3.iterator, StorageLevel.MEMORY_ONLY) Thread.sleep(100) assert(store.get("list2") != None, "list2 was not in store") assert(store.get("list2").get.size == 2) @@ -204,7 +204,7 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT assert(store.get("list2") != None, "list2 was not in store") assert(store.get("list2").get.size == 2) // At this point list2 was gotten last, so LRU will getSingle rid of list3 - store.put("list1", list1.iterator, StorageLevel.MEMORY_ONLY_DESER) + store.put("list1", list1.iterator, StorageLevel.MEMORY_ONLY) Thread.sleep(100) assert(store.get("list1") != None, "list1 was not in store") assert(store.get("list1").get.size == 2) @@ -220,8 +220,8 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT val list3 = List(new Array[Byte](200), new Array[Byte](200)) val list4 = List(new Array[Byte](200), new Array[Byte](200)) // First store list1 and list2, both in memory, and list3, on disk only - store.put("list1", list1.iterator, StorageLevel.MEMORY_ONLY) - store.put("list2", list2.iterator, StorageLevel.MEMORY_ONLY) + store.put("list1", list1.iterator, StorageLevel.MEMORY_ONLY_SER) + store.put("list2", list2.iterator, StorageLevel.MEMORY_ONLY_SER) store.put("list3", list3.iterator, StorageLevel.DISK_ONLY) Thread.sleep(100) // At this point LRU should not kick in because list3 is only on disk @@ -238,7 +238,7 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT assert(store.get("list3") != None, "list1 was not in store") assert(store.get("list3").get.size === 2) // Now let's add in list4, which uses both disk and memory; list1 should drop out - store.put("list4", list4.iterator, StorageLevel.DISK_AND_MEMORY) + store.put("list4", list4.iterator, StorageLevel.MEMORY_AND_DISK_SER) Thread.sleep(100) assert(store.get("list1") === None, "list1 was in store") assert(store.get("list2") != None, "list3 was not in store") diff --git a/docs/scala-programming-guide.md b/docs/scala-programming-guide.md index 28e7bdd4c9..a370bf3ddc 100644 --- a/docs/scala-programming-guide.md +++ b/docs/scala-programming-guide.md @@ -231,30 +231,30 @@ One of the most important capabilities in Spark is *persisting* (or *caching*) a You can mark an RDD to be persisted using the `persist()` or `cache()` methods on it. The first time it is computed in an action, it will be kept in memory on the nodes. The cache is fault-tolerant -- if any partition of an RDD is lost, it will automatically be recomputed using the transformations that originally created it. -In addition, each RDD can be stored using a different *storage level*, allowing you, for example, to persist the dataset on disk, or persist it in memory but as serialized Java objects (to save space), or even replicate it across nodes. These levels are chosen by passing a [`spark.storage.StorageLevel`]({{HOME_PATH}}api/core/index.html#spark.storage.StorageLevel) object to `persist()`. The `cache()` method is a shorthand for using the default storage level, which is `StorageLevel.MEMORY_ONLY_DESER` (store deserialized objects in memory). The complete set of available storage levels is: +In addition, each RDD can be stored using a different *storage level*, allowing you, for example, to persist the dataset on disk, or persist it in memory but as serialized Java objects (to save space), or even replicate it across nodes. These levels are chosen by passing a [`spark.storage.StorageLevel`]({{HOME_PATH}}api/core/index.html#spark.storage.StorageLevel) object to `persist()`. The `cache()` method is a shorthand for using the default storage level, which is `StorageLevel.MEMORY_ONLY` (store deserialized objects in memory). The complete set of available storage levels is: <table class="table"> -<tr><th style="width:30%">Storage Level</th><th>Meaning</th></tr> +<tr><th style="width:23%">Storage Level</th><th>Meaning</th></tr> <tr> - <td> MEMORY_ONLY_DESER </td> + <td> MEMORY_ONLY </td> <td> Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory, some partitions will not be cached and will be recomputed on the fly each time they're needed. This is the default level. </td> </tr> <tr> - <td> DISK_AND_MEMORY_DESER </td> + <td> MEMORY_AND_DISK </td> <td> Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory, store the partitions that don't fit on disk, and read them from there when they're needed. </td> </tr> <tr> - <td> MEMORY_ONLY </td> - <td> Store RDD as <i>serialized</i> Java objects (that is, one byte array per partition). + <td> MEMORY_ONLY_SER </td> + <td> Store RDD as <i>serialized</i> Java objects (one byte array per partition). This is generally more space-efficient than deserialized objects, especially when using a <a href="{{HOME_PATH}}tuning.html">fast serializer</a>, but more CPU-intensive to read. </td> </tr> <tr> - <td> DISK_AND_MEMORY </td> - <td> Similar to MEMORY_ONLY, but spill partitions that don't fit in memory to disk instead of recomputing them + <td> MEMORY_AND_DISK_SER </td> + <td> Similar to MEMORY_ONLY_SER, but spill partitions that don't fit in memory to disk instead of recomputing them on the fly each time they're needed. </td> </tr> <tr> @@ -262,8 +262,8 @@ In addition, each RDD can be stored using a different *storage level*, allowing <td> Store the RDD partitions only on disk. </td> </tr> <tr> - <td> MEMORY_ONLY_DESER_2 / DISK_AND_MEMORY_DESER_2 / MEMORY_ONLY_2 / DISK_ONLY_2 / DISK_AND_MEMORY_2 </td> - <td> Same as the levels above, but replicate each partition on two nodes. </td> + <td> MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc. </td> + <td> Same as the levels above, but replicate each partition on two cluster nodes. </td> </tr> </table> @@ -272,9 +272,9 @@ In addition, each RDD can be stored using a different *storage level*, allowing Spark's storage levels are meant to provide different tradeoffs between memory usage and CPU efficiency. We recommend going through the following process to select one: -* If your RDDs fit comfortably with the default storage level (`MEMORY_ONLY_DESER`), leave them that way. This is the most +* If your RDDs fit comfortably with the default storage level (`MEMORY_ONLY`), leave them that way. This is the most CPU-efficient option, allowing operations on the RDDs to run as fast as possible. -* If not, try using `MEMORY_ONLY` and [selecting a fast serialization library]({{HOME_PATH}}tuning.html) to make the objects +* If not, try using `MEMORY_ONLY_SER` and [selecting a fast serialization library]({{HOME_PATH}}tuning.html) to make the objects much more space-efficient, but still reasonably fast to access. * Don't spill to disk unless the functions that computed your datasets are expensive, or they filter a large amount of the data. Otherwise, recomputing a partition is about as fast as reading it from disk. |