You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by andrewor14 <gi...@git.apache.org> on 2014/07/31 05:36:05 UTC
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
GitHub user andrewor14 opened a pull request:
https://github.com/apache/spark/pull/1679
[SPARK-2316] Avoid O(blocks) operations in listeners
The existing code in `StorageUtils` is not the most efficient. Every time we want to update an `RDDInfo` we end up iterating through all blocks on all block managers just to discard most of them. The symptoms manifest themselves in the bountiful UI bugs observed in the wild. Many of these bugs are caused by the slow consumption of events in `LiveListenerBus`, which frequently leads to the event queue overflowing and `SparkListenerEvent`s being dropped on the floor. The changes made in this PR avoid this by first filtering out only the blocks relevant to us before computing storage information from them.
It's worth a mention that this corner of the Spark code is also not very well-tested at all. The bulk of the changes in this PR is actually test cases for various logic in `StorageUtils.scala`. These will eventually be extended to cover the various listeners that constitute the `SparkUI`.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/andrewor14/spark fix-drop-events
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/1679.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1679
----
commit 53af15d25e19b8b63bcad035e3149e2920943561
Author: Andrew Or <an...@gmail.com>
Date: 2014-07-30T23:20:40Z
Refactor StorageStatus + add a bunch of tests
This commit refactors storage status to keep around a set of RDD
IDs which have blocks stored in the status' block manager. The
purpose is such that we don't have to linearly scan through every
single storage status' blocks if it doesn't even contain blocks
for the RDD we're interested in in the first place.
This commit also adds a bunch of tests for StorageStatus and
StorageUtils methods. There were previously a few minor bugs in
StorageUtils.blockLocationsFromStorageStatus and
StorageUtils.filterStorageStatusByRDD that are now fixed and tested.
Going forward, we need to first cleanup the method signatures to
reflect what they actually do. Then we will make things more
efficient now that we've set the stage.
commit 41fa50df1fc520802905b2f716b2008004c7c79d
Author: Andrew Or <an...@gmail.com>
Date: 2014-07-31T01:51:56Z
Add a legacy constructor for StorageStatus
This just makes it easier to create one with a source of blocks.
commit 7b2c4aae86c784e117809fd857c31a3a402dd958
Author: Andrew Or <an...@gmail.com>
Date: 2014-07-31T02:18:52Z
Rewrite blockLocationsFromStorageStatus + clean up method signatures
The existing implementation of blockLocationFromStorageStatus relies
on a groupBy, which is somewhat expensive. The new code creates a map
from the get go and adds the block locations by iterating through the
storage statuses' blocks.
This commit also cleans up StorageUtils method signatures by removing
unnecessary methods and renaming others with long-winded names.
commit 8e91921983fc385896d9946303debd9e77652d6c
Author: Andrew Or <an...@gmail.com>
Date: 2014-07-31T03:21:31Z
Iterate through a filtered set of blocks when updating RDDInfo
This particular commit is the whole point of this PR. In the existing
code we unconditionally iterate through all blocks in all block managers
whenever we want to update an RDDInfo. Now, we filter out only the
blocks of interest to us in advance, so we don't end up constructing
a huge map and doing a groupBy on it.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15725945
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,263 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * A common consumption pattern is to access only the blocks that belong to a specific RDD.
+ * For this use case, we should avoid linearly scanning through all the blocks, which could
+ * be expensive if there are thousands of blocks on each block manager. Thus, we need to store
+ * RDD blocks and non-RDD blocks separately. In particular, we store RDD blocks in a map
+ * indexed by RDD IDs, so we can filter out the blocks of interest quickly.
- def memUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ * These collections should only be mutated through the add/update/removeBlock methods.
+ */
+ private val _rddBlocks = new mutable.HashMap[Int, mutable.Map[BlockId, BlockStatus]]
+ private val _nonRddBlocks = new mutable.HashMap[BlockId, BlockStatus]
- def diskUsed = blocks.values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * A map of storage information associated with each RDD.
+ *
+ * The key is the ID of the RDD, and the value is a 4-tuple of the following:
+ * (size in memory, size on disk, size in tachyon, storage level)
+ *
+ * This is updated incrementally on each block added, updated or removed, so as to avoid
+ * linearly scanning through all the blocks within an RDD if we're only interested in a
+ * given RDD's storage information.
+ */
+ private val _rddStorageInfo = new mutable.HashMap[Int, (Long, Long, Long, StorageLevel)]
- def diskUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Instantiate a StorageStatus with the given initial blocks. This essentially makes a copy of
+ * the original blocks map such that the fate of this storage status is not tied to the source.
+ */
+ def this(bmid: BlockManagerId, maxMem: Long, initialBlocks: Map[BlockId, BlockStatus]) {
+ this(bmid, maxMem)
+ initialBlocks.foreach { case (bid, bstatus) => addBlock(bid, bstatus) }
+ }
- def memRemaining: Long = maxMem - memUsed
+ /**
+ * Return the blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * contains, get, and size.
+ */
+ def blocks: Map[BlockId, BlockStatus] = _nonRddBlocks ++ rddBlocks
- def rddBlocks = blocks.collect { case (rdd: RDDBlockId, status) => (rdd, status) }
-}
+ /**
+ * Return the RDD blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * getting the memory, disk, and off-heap memory sizes occupied by this RDD.
+ */
+ def rddBlocks: Map[BlockId, BlockStatus] = _rddBlocks.flatMap { case (_, blocks) => blocks }
-/** Helper methods for storage-related objects. */
-private[spark] object StorageUtils {
+ /** Return the blocks that belong to the given RDD stored in this block manager. */
+ def rddBlocksById(rddId: Int): Map[BlockId, BlockStatus] = {
+ _rddBlocks.get(rddId).getOrElse(Map.empty)
+ }
+
+ /** Add the given block to this storage status. If it already exists, overwrite it. */
+ def addBlock(blockId: BlockId, blockStatus: BlockStatus): Unit = {
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ // Update the storage info of the RDD, keeping track of any existing status for this block
+ val oldBlockStatus = getBlock(blockId).getOrElse(BlockStatus.empty)
+ val changeInMem = blockStatus.memSize - oldBlockStatus.memSize
--- End diff --
should we add something like `BlockStatus.subtract` and `BlockStatus.add` here? It might not totally make sense since not all of the fields in BlockStatus are numeric.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50952490
If you fix the `sum` thing I think this is ready to go
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50951768
@shivaram did your thing finish alright?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP][SPARK-2316] Avoid O(blocks) operations i...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50855850
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17659/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50825453
QA results for PR 1679:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17608/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by concretevitamin <gi...@git.apache.org>.
Github user concretevitamin commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50947806
Yeah, I was asking how many default listeners were in the default set. This
number can somewhat hint at the behavior when users add their own
listeners.
On Friday, August 1, 2014, andrewor14 <no...@github.com> wrote:
> @concretevitamin <https://github.com/concretevitamin> What do you mean? I
> am using the default set of listeners attached to the SparkUI.
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/1679#issuecomment-50946929>.
>
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50935078
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17699/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50793747
I still need to do some benchmarking, but this seems like the most expensive operation the listeners have to carry out, especially when there are thousands of blocks on each executor.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50903178
Hm, test this please
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50845389
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17636/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50819549
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17608/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50819002
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17607/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50950885
With the latest commit, you can refresh the executors page without waiting for 5 seconds!! (assuming 100k total blocks)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50955955
I've merged this, thanks. It could be worth back porting into branch-1.0 as well, but I didn't do that yet.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15725862
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,263 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * A common consumption pattern is to access only the blocks that belong to a specific RDD.
+ * For this use case, we should avoid linearly scanning through all the blocks, which could
+ * be expensive if there are thousands of blocks on each block manager. Thus, we need to store
+ * RDD blocks and non-RDD blocks separately. In particular, we store RDD blocks in a map
+ * indexed by RDD IDs, so we can filter out the blocks of interest quickly.
- def memUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ * These collections should only be mutated through the add/update/removeBlock methods.
+ */
+ private val _rddBlocks = new mutable.HashMap[Int, mutable.Map[BlockId, BlockStatus]]
+ private val _nonRddBlocks = new mutable.HashMap[BlockId, BlockStatus]
- def diskUsed = blocks.values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * A map of storage information associated with each RDD.
+ *
+ * The key is the ID of the RDD, and the value is a 4-tuple of the following:
+ * (size in memory, size on disk, size in tachyon, storage level)
+ *
+ * This is updated incrementally on each block added, updated or removed, so as to avoid
+ * linearly scanning through all the blocks within an RDD if we're only interested in a
+ * given RDD's storage information.
+ */
+ private val _rddStorageInfo = new mutable.HashMap[Int, (Long, Long, Long, StorageLevel)]
- def diskUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Instantiate a StorageStatus with the given initial blocks. This essentially makes a copy of
--- End diff --
I think you can just say "This will leave the supplied map unmodified.".
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50806389
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17596/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50955077
LGTM pending tests.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50940127
QA results for PR 1679:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17699/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP][SPARK-2316] Avoid O(blocks) operations i...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50857283
I did some benchmarking by running the following job 100 times one immediately after another. Each job launches a many short-lived tasks, each of which persists a single block. The minimality of each task allows the listener bus to keep posting events very quickly while placing a lot of stress on the listeners on consuming the events.
```
sc.parallelize(1 to 20000, 100).persist().count()
```
**Before:** The max queue length observed reaches 10000 at around the 65th job, and finally reaches 16730 after the last job. Before this PR, this is enough to cause the queue to start dropping events. The average time spent in `StorageUtils.updateRddInfo` (this was renamed) is 176.25ms.
**After:** The max queue length never went above 130, and the average time spent in `StorageUtils.updateRddInfo` is 15.47ms, more than 10 times faster than before.
The dark side of the story (there is always a dark side), however, is that this improvement is only observed for RDDs with not too many partitions. Although the new code iterates through only a few RDDs' blocks instead of all RDD blocks known to mankind, it is still slow if say a single RDD contains all the blocks, in which case we still have to iterate through all the RDD blocks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50847382
QA results for PR 1679:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17636/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/1679
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15725915
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,263 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * A common consumption pattern is to access only the blocks that belong to a specific RDD.
+ * For this use case, we should avoid linearly scanning through all the blocks, which could
+ * be expensive if there are thousands of blocks on each block manager. Thus, we need to store
+ * RDD blocks and non-RDD blocks separately. In particular, we store RDD blocks in a map
+ * indexed by RDD IDs, so we can filter out the blocks of interest quickly.
- def memUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ * These collections should only be mutated through the add/update/removeBlock methods.
+ */
+ private val _rddBlocks = new mutable.HashMap[Int, mutable.Map[BlockId, BlockStatus]]
+ private val _nonRddBlocks = new mutable.HashMap[BlockId, BlockStatus]
- def diskUsed = blocks.values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * A map of storage information associated with each RDD.
+ *
+ * The key is the ID of the RDD, and the value is a 4-tuple of the following:
+ * (size in memory, size on disk, size in tachyon, storage level)
+ *
+ * This is updated incrementally on each block added, updated or removed, so as to avoid
+ * linearly scanning through all the blocks within an RDD if we're only interested in a
+ * given RDD's storage information.
+ */
+ private val _rddStorageInfo = new mutable.HashMap[Int, (Long, Long, Long, StorageLevel)]
- def diskUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Instantiate a StorageStatus with the given initial blocks. This essentially makes a copy of
+ * the original blocks map such that the fate of this storage status is not tied to the source.
+ */
+ def this(bmid: BlockManagerId, maxMem: Long, initialBlocks: Map[BlockId, BlockStatus]) {
+ this(bmid, maxMem)
+ initialBlocks.foreach { case (bid, bstatus) => addBlock(bid, bstatus) }
+ }
- def memRemaining: Long = maxMem - memUsed
+ /**
+ * Return the blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * contains, get, and size.
+ */
+ def blocks: Map[BlockId, BlockStatus] = _nonRddBlocks ++ rddBlocks
--- End diff --
this is only used internally by `memUsed`, `diskUsed`, etc. Should we make those more efficient also now that we can just add the totals from each rdd (and then add up other blocks)?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50903384
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17679/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50954833
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17755/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50824930
QA results for PR 1679:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17607/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by aarondav <gi...@git.apache.org>.
Github user aarondav commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50714796
Were you able to test the performance characteristics of this versus the old stuff? Was this indeed the main cause of the live listener bus overflowing, or is that still a problem?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15726191
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,263 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * A common consumption pattern is to access only the blocks that belong to a specific RDD.
+ * For this use case, we should avoid linearly scanning through all the blocks, which could
+ * be expensive if there are thousands of blocks on each block manager. Thus, we need to store
+ * RDD blocks and non-RDD blocks separately. In particular, we store RDD blocks in a map
+ * indexed by RDD IDs, so we can filter out the blocks of interest quickly.
- def memUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ * These collections should only be mutated through the add/update/removeBlock methods.
+ */
+ private val _rddBlocks = new mutable.HashMap[Int, mutable.Map[BlockId, BlockStatus]]
+ private val _nonRddBlocks = new mutable.HashMap[BlockId, BlockStatus]
- def diskUsed = blocks.values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * A map of storage information associated with each RDD.
+ *
+ * The key is the ID of the RDD, and the value is a 4-tuple of the following:
+ * (size in memory, size on disk, size in tachyon, storage level)
+ *
+ * This is updated incrementally on each block added, updated or removed, so as to avoid
+ * linearly scanning through all the blocks within an RDD if we're only interested in a
+ * given RDD's storage information.
+ */
+ private val _rddStorageInfo = new mutable.HashMap[Int, (Long, Long, Long, StorageLevel)]
- def diskUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Instantiate a StorageStatus with the given initial blocks. This essentially makes a copy of
+ * the original blocks map such that the fate of this storage status is not tied to the source.
+ */
+ def this(bmid: BlockManagerId, maxMem: Long, initialBlocks: Map[BlockId, BlockStatus]) {
+ this(bmid, maxMem)
+ initialBlocks.foreach { case (bid, bstatus) => addBlock(bid, bstatus) }
+ }
- def memRemaining: Long = maxMem - memUsed
+ /**
+ * Return the blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * contains, get, and size.
+ */
+ def blocks: Map[BlockId, BlockStatus] = _nonRddBlocks ++ rddBlocks
- def rddBlocks = blocks.collect { case (rdd: RDDBlockId, status) => (rdd, status) }
-}
+ /**
+ * Return the RDD blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * getting the memory, disk, and off-heap memory sizes occupied by this RDD.
+ */
+ def rddBlocks: Map[BlockId, BlockStatus] = _rddBlocks.flatMap { case (_, blocks) => blocks }
-/** Helper methods for storage-related objects. */
-private[spark] object StorageUtils {
+ /** Return the blocks that belong to the given RDD stored in this block manager. */
+ def rddBlocksById(rddId: Int): Map[BlockId, BlockStatus] = {
+ _rddBlocks.get(rddId).getOrElse(Map.empty)
+ }
+
+ /** Add the given block to this storage status. If it already exists, overwrite it. */
+ def addBlock(blockId: BlockId, blockStatus: BlockStatus): Unit = {
--- End diff --
Perhaps we should make all of the mutation methods `private[spark]`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15725971
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,263 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * A common consumption pattern is to access only the blocks that belong to a specific RDD.
+ * For this use case, we should avoid linearly scanning through all the blocks, which could
+ * be expensive if there are thousands of blocks on each block manager. Thus, we need to store
+ * RDD blocks and non-RDD blocks separately. In particular, we store RDD blocks in a map
+ * indexed by RDD IDs, so we can filter out the blocks of interest quickly.
- def memUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ * These collections should only be mutated through the add/update/removeBlock methods.
+ */
+ private val _rddBlocks = new mutable.HashMap[Int, mutable.Map[BlockId, BlockStatus]]
+ private val _nonRddBlocks = new mutable.HashMap[BlockId, BlockStatus]
- def diskUsed = blocks.values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * A map of storage information associated with each RDD.
+ *
+ * The key is the ID of the RDD, and the value is a 4-tuple of the following:
+ * (size in memory, size on disk, size in tachyon, storage level)
+ *
+ * This is updated incrementally on each block added, updated or removed, so as to avoid
+ * linearly scanning through all the blocks within an RDD if we're only interested in a
+ * given RDD's storage information.
+ */
+ private val _rddStorageInfo = new mutable.HashMap[Int, (Long, Long, Long, StorageLevel)]
- def diskUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Instantiate a StorageStatus with the given initial blocks. This essentially makes a copy of
+ * the original blocks map such that the fate of this storage status is not tied to the source.
+ */
+ def this(bmid: BlockManagerId, maxMem: Long, initialBlocks: Map[BlockId, BlockStatus]) {
+ this(bmid, maxMem)
+ initialBlocks.foreach { case (bid, bstatus) => addBlock(bid, bstatus) }
+ }
- def memRemaining: Long = maxMem - memUsed
+ /**
+ * Return the blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * contains, get, and size.
+ */
+ def blocks: Map[BlockId, BlockStatus] = _nonRddBlocks ++ rddBlocks
- def rddBlocks = blocks.collect { case (rdd: RDDBlockId, status) => (rdd, status) }
-}
+ /**
+ * Return the RDD blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * getting the memory, disk, and off-heap memory sizes occupied by this RDD.
+ */
+ def rddBlocks: Map[BlockId, BlockStatus] = _rddBlocks.flatMap { case (_, blocks) => blocks }
-/** Helper methods for storage-related objects. */
-private[spark] object StorageUtils {
+ /** Return the blocks that belong to the given RDD stored in this block manager. */
+ def rddBlocksById(rddId: Int): Map[BlockId, BlockStatus] = {
+ _rddBlocks.get(rddId).getOrElse(Map.empty)
+ }
+
+ /** Add the given block to this storage status. If it already exists, overwrite it. */
+ def addBlock(blockId: BlockId, blockStatus: BlockStatus): Unit = {
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ // Update the storage info of the RDD, keeping track of any existing status for this block
+ val oldBlockStatus = getBlock(blockId).getOrElse(BlockStatus.empty)
+ val changeInMem = blockStatus.memSize - oldBlockStatus.memSize
+ val changeInDisk = blockStatus.diskSize - oldBlockStatus.diskSize
+ val changeInTachyon = blockStatus.tachyonSize - oldBlockStatus.tachyonSize
+ val level = blockStatus.storageLevel
+ updateRddStorageInfo(rddId, changeInMem, changeInDisk, changeInTachyon, level)
+ // Actually add the block itself
+ _rddBlocks.getOrElseUpdate(rddId, new mutable.HashMap)(blockId) = blockStatus
+ case _ =>
+ _nonRddBlocks(blockId) = blockStatus
+ }
+ }
+
+ /** Update the given block in this storage status. If it doesn't already exist, add it. */
+ def updateBlock(blockId: BlockId, blockStatus: BlockStatus): Unit = {
+ addBlock(blockId, blockStatus)
+ }
+
+ /** Remove the given block from this storage status. */
+ def removeBlock(blockId: BlockId): Option[BlockStatus] = {
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ // Update the storage info of the RDD if the block to remove exists
+ getBlock(blockId).foreach { s =>
+ updateRddStorageInfo(rddId, -s.memSize, -s.diskSize, -s.tachyonSize, StorageLevel.NONE)
+ }
+ // Actually remove the block, if it exists
+ if (_rddBlocks.contains(rddId)) {
--- End diff --
Is there a case where we'd expect this not to be present in this map?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15726507
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,258 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * We store RDD blocks and non-RDD blocks separately to allow quick retrievals of RDD blocks.
+ * These collections should only be mutated through the add/update/removeBlock methods.
+ */
+ private val _rddBlocks = new mutable.HashMap[Int, mutable.Map[BlockId, BlockStatus]]
+ private val _nonRddBlocks = new mutable.HashMap[BlockId, BlockStatus]
- def memUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Storage information of the blocks that entails memory, disk, and off-heap memory usage.
+ *
+ * As with the block maps, we store the storage information separately for RDD blocks and
+ * non-RDD blocks for the same reason. In particular, RDD storage information is stored
+ * in a map indexed by the RDD ID to the following 4-tuple:
+ *
+ * (memory size, disk size, off-heap size, storage level)
+ *
+ * We assume that all the blocks that belong to the same RDD have the same storage level.
+ * This field is not relevant to non-RDD blocks, however, so the storage information for
+ * non-RDD blocks contains only the first 3 fields (in the same order).
+ */
+ private val _rddStorageInfo = new mutable.HashMap[Int, (Long, Long, Long, StorageLevel)]
+ private var _nonRddStorageInfo: (Long, Long, Long) = (0L, 0L, 0L)
- def diskUsed = blocks.values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /** Create a storage status with an initial set of blocks, leaving the source unmodified. */
+ def this(bmid: BlockManagerId, maxMem: Long, initialBlocks: Map[BlockId, BlockStatus]) {
+ this(bmid, maxMem)
+ initialBlocks.foreach { case (bid, bstatus) => addBlock(bid, bstatus) }
+ }
- def diskUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Return the blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * contains, get, and size.
+ */
+ def blocks: Map[BlockId, BlockStatus] = _nonRddBlocks ++ rddBlocks
- def memRemaining: Long = maxMem - memUsed
+ /**
+ * Return the RDD blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * getting the memory, disk, and off-heap memory sizes occupied by this RDD.
+ */
+ def rddBlocks: Map[BlockId, BlockStatus] = _rddBlocks.flatMap { case (_, blocks) => blocks }
- def rddBlocks = blocks.collect { case (rdd: RDDBlockId, status) => (rdd, status) }
-}
+ /** Return the blocks that belong to the given RDD stored in this block manager. */
+ def rddBlocksById(rddId: Int): Map[BlockId, BlockStatus] = {
+ _rddBlocks.get(rddId).getOrElse(Map.empty)
+ }
-/** Helper methods for storage-related objects. */
-private[spark] object StorageUtils {
+ /** Add the given block to this storage status. If it already exists, overwrite it. */
+ private[spark] def addBlock(blockId: BlockId, blockStatus: BlockStatus): Unit = {
+ updateStorageInfo(blockId, blockStatus)
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ _rddBlocks.getOrElseUpdate(rddId, new mutable.HashMap)(blockId) = blockStatus
+ case _ =>
+ _nonRddBlocks(blockId) = blockStatus
+ }
+ }
+
+ /** Update the given block in this storage status. If it doesn't already exist, add it. */
+ private[spark] def updateBlock(blockId: BlockId, blockStatus: BlockStatus): Unit = {
+ addBlock(blockId, blockStatus)
+ }
+
+ /** Remove the given block from this storage status. */
+ private[spark] def removeBlock(blockId: BlockId): Option[BlockStatus] = {
+ updateStorageInfo(blockId, BlockStatus.empty)
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ // Actually remove the block, if it exists
+ if (_rddBlocks.contains(rddId)) {
+ val removed = _rddBlocks(rddId).remove(blockId)
+ // If the given RDD has no more blocks left, remove the RDD
+ if (_rddBlocks(rddId).isEmpty) {
+ _rddBlocks.remove(rddId)
+ }
+ removed
+ } else {
+ None
+ }
+ case _ =>
+ _nonRddBlocks.remove(blockId)
+ }
+ }
/**
- * Returns basic information of all RDDs persisted in the given SparkContext. This does not
- * include storage information.
+ * Return whether the given block is stored in this block manager in O(1) time.
+ * Note that this is much faster than `this.blocks.contains`, which is O(blocks) time.
*/
- def rddInfoFromSparkContext(sc: SparkContext): Array[RDDInfo] = {
- sc.persistentRdds.values.map { rdd =>
- val rddName = Option(rdd.name).getOrElse(rdd.id.toString)
- val rddNumPartitions = rdd.partitions.size
- val rddStorageLevel = rdd.getStorageLevel
- val rddInfo = new RDDInfo(rdd.id, rddName, rddNumPartitions, rddStorageLevel)
- rddInfo
- }.toArray
+ def containsBlock(blockId: BlockId): Boolean = {
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ _rddBlocks.get(rddId).exists(_.contains(blockId))
+ case _ =>
+ _nonRddBlocks.contains(blockId)
+ }
}
- /** Returns storage information of all RDDs persisted in the given SparkContext. */
- def rddInfoFromStorageStatus(
- storageStatuses: Seq[StorageStatus],
- sc: SparkContext): Array[RDDInfo] = {
- rddInfoFromStorageStatus(storageStatuses, rddInfoFromSparkContext(sc))
+ /**
+ * Return the given block stored in this block manager in O(1) time.
+ * Note that this is much faster than `this.blocks.get`, which is O(blocks) time.
+ */
+ def getBlock(blockId: BlockId): Option[BlockStatus] = {
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ _rddBlocks.get(rddId).map(_.get(blockId)).flatten
+ case _ =>
+ _nonRddBlocks.get(blockId)
+ }
}
- /** Returns storage information of all RDDs in the given list. */
- def rddInfoFromStorageStatus(
- storageStatuses: Seq[StorageStatus],
- rddInfos: Seq[RDDInfo],
- updatedBlocks: Seq[(BlockId, BlockStatus)] = Seq.empty): Array[RDDInfo] = {
-
- // Mapping from a block ID -> its status
- val blockMap = mutable.Map(storageStatuses.flatMap(_.rddBlocks): _*)
-
- // Record updated blocks, if any
- updatedBlocks
- .collect { case (id: RDDBlockId, status) => (id, status) }
- .foreach { case (id, status) => blockMap(id) = status }
-
- // Mapping from RDD ID -> an array of associated BlockStatuses
- val rddBlockMap = blockMap
- .groupBy { case (k, _) => k.rddId }
- .mapValues(_.values.toArray)
-
- // Mapping from RDD ID -> the associated RDDInfo (with potentially outdated storage information)
- val rddInfoMap = rddInfos.map { info => (info.id, info) }.toMap
-
- val rddStorageInfos = rddBlockMap.flatMap { case (rddId, blocks) =>
- // Add up memory, disk and Tachyon sizes
- val persistedBlocks =
- blocks.filter { status => status.memSize + status.diskSize + status.tachyonSize > 0 }
- val _storageLevel =
- if (persistedBlocks.length > 0) persistedBlocks(0).storageLevel else StorageLevel.NONE
- val memSize = persistedBlocks.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
- val diskSize = persistedBlocks.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
- val tachyonSize = persistedBlocks.map(_.tachyonSize).reduceOption(_ + _).getOrElse(0L)
- rddInfoMap.get(rddId).map { rddInfo =>
- rddInfo.storageLevel = _storageLevel
- rddInfo.numCachedPartitions = persistedBlocks.length
- rddInfo.memSize = memSize
- rddInfo.diskSize = diskSize
- rddInfo.tachyonSize = tachyonSize
- rddInfo
- }
- }.toArray
+ /**
+ * Return the number of blocks stored in this block manager in O(RDDs) time.
+ * Note that this is much faster than `this.blocks.size`, which is O(blocks) time.
+ */
+ def numBlocks: Int = _nonRddBlocks.size + numRddBlocks
+
+ /**
+ * Return the number of RDD blocks stored in this block manager in O(RDDs) time.
+ * Note that this is much faster than `this.rddBlocks.size`, which is O(RDD blocks) time.
+ */
+ def numRddBlocks: Int = _rddBlocks.values.map(_.size).fold(0)(_ + _)
- scala.util.Sorting.quickSort(rddStorageInfos)
- rddStorageInfos
+ /**
+ * Return the number of blocks that belong to the given RDD in O(1) time.
+ * Note that this is much faster than `this.rddBlocksById(rddId).size`, which is
+ * O(blocks in this RDD) time.
+ */
+ def numRddBlocksById(rddId: Int): Int = _rddBlocks.get(rddId).map(_.size).getOrElse(0)
+
+ /** Return the memory remaining in this block manager. */
+ def memRemaining: Long = maxMem - memUsed
+
+ /** Return the memory used by this block manager. */
+ def memUsed: Long =
+ _nonRddStorageInfo._1 + _rddBlocks.keys.toSeq.map(memUsedByRdd).fold(0L)(_ + _)
--- End diff --
We can replace all of these aggregations with `sum`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50711227
QA results for PR 1679:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17547/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by concretevitamin <gi...@git.apache.org>.
Github user concretevitamin commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50827351
This patch is much appreciated -- thanks for working on this!
On Thu, Jul 31, 2014 at 3:02 PM, Apache Spark QA <no...@github.com>
wrote:
> QA results for PR 1679:
> - This patch PASSES unit tests.
> - This patch merges cleanly
> - This patch adds the following public classes (experimental):
> class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
>
> For more information see test ouptut:
>
> https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17607/consoleFull
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/1679#issuecomment-50824930>.
>
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50709239
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17547/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50954718
Great to know @shivaram. Thanks for testing.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15726131
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,263 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * A common consumption pattern is to access only the blocks that belong to a specific RDD.
+ * For this use case, we should avoid linearly scanning through all the blocks, which could
+ * be expensive if there are thousands of blocks on each block manager. Thus, we need to store
+ * RDD blocks and non-RDD blocks separately. In particular, we store RDD blocks in a map
+ * indexed by RDD IDs, so we can filter out the blocks of interest quickly.
- def memUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ * These collections should only be mutated through the add/update/removeBlock methods.
+ */
+ private val _rddBlocks = new mutable.HashMap[Int, mutable.Map[BlockId, BlockStatus]]
+ private val _nonRddBlocks = new mutable.HashMap[BlockId, BlockStatus]
- def diskUsed = blocks.values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * A map of storage information associated with each RDD.
+ *
+ * The key is the ID of the RDD, and the value is a 4-tuple of the following:
+ * (size in memory, size on disk, size in tachyon, storage level)
+ *
+ * This is updated incrementally on each block added, updated or removed, so as to avoid
+ * linearly scanning through all the blocks within an RDD if we're only interested in a
+ * given RDD's storage information.
+ */
+ private val _rddStorageInfo = new mutable.HashMap[Int, (Long, Long, Long, StorageLevel)]
- def diskUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Instantiate a StorageStatus with the given initial blocks. This essentially makes a copy of
+ * the original blocks map such that the fate of this storage status is not tied to the source.
+ */
+ def this(bmid: BlockManagerId, maxMem: Long, initialBlocks: Map[BlockId, BlockStatus]) {
+ this(bmid, maxMem)
+ initialBlocks.foreach { case (bid, bstatus) => addBlock(bid, bstatus) }
+ }
- def memRemaining: Long = maxMem - memUsed
+ /**
+ * Return the blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * contains, get, and size.
+ */
+ def blocks: Map[BlockId, BlockStatus] = _nonRddBlocks ++ rddBlocks
--- End diff --
Coming next commit. Look forward to it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50909399
QA results for PR 1679:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17679/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50946913
One thing that is still slightly worrying is that every time we refresh the executors page, there is still a noticeable delay if we have persisted tens of thousands of blocks. This is because we still iterate through all the blocks there to find out how much memory each executor currently uses. We can fix this by incrementally updating memory and disk usage for non-RDD blocks similarly to what we have done in this PR for RDD blocks. However, this issue does not affect the event consumption rate and I may fix this separately.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50954735
Jenkins, test this please.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50950925
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17731/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by concretevitamin <gi...@git.apache.org>.
Github user concretevitamin commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50946639
How many listeners are used in these benchmarks?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15726421
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,263 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * A common consumption pattern is to access only the blocks that belong to a specific RDD.
+ * For this use case, we should avoid linearly scanning through all the blocks, which could
+ * be expensive if there are thousands of blocks on each block manager. Thus, we need to store
+ * RDD blocks and non-RDD blocks separately. In particular, we store RDD blocks in a map
+ * indexed by RDD IDs, so we can filter out the blocks of interest quickly.
- def memUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ * These collections should only be mutated through the add/update/removeBlock methods.
+ */
+ private val _rddBlocks = new mutable.HashMap[Int, mutable.Map[BlockId, BlockStatus]]
+ private val _nonRddBlocks = new mutable.HashMap[BlockId, BlockStatus]
- def diskUsed = blocks.values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * A map of storage information associated with each RDD.
+ *
+ * The key is the ID of the RDD, and the value is a 4-tuple of the following:
+ * (size in memory, size on disk, size in tachyon, storage level)
+ *
+ * This is updated incrementally on each block added, updated or removed, so as to avoid
+ * linearly scanning through all the blocks within an RDD if we're only interested in a
+ * given RDD's storage information.
+ */
+ private val _rddStorageInfo = new mutable.HashMap[Int, (Long, Long, Long, StorageLevel)]
- def diskUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Instantiate a StorageStatus with the given initial blocks. This essentially makes a copy of
+ * the original blocks map such that the fate of this storage status is not tied to the source.
+ */
+ def this(bmid: BlockManagerId, maxMem: Long, initialBlocks: Map[BlockId, BlockStatus]) {
+ this(bmid, maxMem)
+ initialBlocks.foreach { case (bid, bstatus) => addBlock(bid, bstatus) }
+ }
- def memRemaining: Long = maxMem - memUsed
+ /**
+ * Return the blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * contains, get, and size.
+ */
+ def blocks: Map[BlockId, BlockStatus] = _nonRddBlocks ++ rddBlocks
- def rddBlocks = blocks.collect { case (rdd: RDDBlockId, status) => (rdd, status) }
-}
+ /**
+ * Return the RDD blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * getting the memory, disk, and off-heap memory sizes occupied by this RDD.
+ */
+ def rddBlocks: Map[BlockId, BlockStatus] = _rddBlocks.flatMap { case (_, blocks) => blocks }
-/** Helper methods for storage-related objects. */
-private[spark] object StorageUtils {
+ /** Return the blocks that belong to the given RDD stored in this block manager. */
+ def rddBlocksById(rddId: Int): Map[BlockId, BlockStatus] = {
+ _rddBlocks.get(rddId).getOrElse(Map.empty)
+ }
+
+ /** Add the given block to this storage status. If it already exists, overwrite it. */
+ def addBlock(blockId: BlockId, blockStatus: BlockStatus): Unit = {
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ // Update the storage info of the RDD, keeping track of any existing status for this block
+ val oldBlockStatus = getBlock(blockId).getOrElse(BlockStatus.empty)
+ val changeInMem = blockStatus.memSize - oldBlockStatus.memSize
+ val changeInDisk = blockStatus.diskSize - oldBlockStatus.diskSize
+ val changeInTachyon = blockStatus.tachyonSize - oldBlockStatus.tachyonSize
+ val level = blockStatus.storageLevel
+ updateRddStorageInfo(rddId, changeInMem, changeInDisk, changeInTachyon, level)
+ // Actually add the block itself
+ _rddBlocks.getOrElseUpdate(rddId, new mutable.HashMap)(blockId) = blockStatus
+ case _ =>
+ _nonRddBlocks(blockId) = blockStatus
+ }
+ }
+
+ /** Update the given block in this storage status. If it doesn't already exist, add it. */
+ def updateBlock(blockId: BlockId, blockStatus: BlockStatus): Unit = {
+ addBlock(blockId, blockStatus)
+ }
+
+ /** Remove the given block from this storage status. */
+ def removeBlock(blockId: BlockId): Option[BlockStatus] = {
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ // Update the storage info of the RDD if the block to remove exists
+ getBlock(blockId).foreach { s =>
+ updateRddStorageInfo(rddId, -s.memSize, -s.diskSize, -s.tachyonSize, StorageLevel.NONE)
+ }
+ // Actually remove the block, if it exists
+ if (_rddBlocks.contains(rddId)) {
--- End diff --
Not really, just that it doesn't throw some key exception if we misuse this
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50952187
QA results for PR 1679:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17731/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50849016
QA tests have started for PR 1679. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17644/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP][SPARK-2316] Avoid O(blocks) operations i...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50851416
QA results for PR 1679:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17644/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by shivaram <gi...@git.apache.org>.
Github user shivaram commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50952987
@pwendell @andrewor14
Yes - the run went fine. I didn't see any listener bus overflows and the UI was fine. Also I used to previously see 1 CPU fully occupied by the StorageStatus stuff -- This time the CPU usage remained lower.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50949346
I think the default set has 5 or 6 listeners, but actually this is slow because one listener is taking up all the time, but not that there are many listeners. The other listeners are actually taking up a trivial amount of time in comparison to process the events.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50859282
QA results for PR 1679:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17659/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50848836
test this please
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50946509
After the latest changes, I did some more benchmarking running the same jobs with more partitions. I ran two experiments with different number of partitions, 1000 and 10000. For each experiment, I ran the following job 10 times, one immediately after another, and recorded the max queue length.
```
sc.parallelize(1 to 20000, numPartitions).persist().count()
```
**Before.** In the experiment with 1000-partition RDDs, the event queue length exceeded 10k at around the 6th job and eventually reached 18019 after the 10th job. These numbers are even higher for 10000-partition RDDs, in which the event queue length already exceeded 10k by the 1st job, and finally reached *193162* after the 10th. Under this (contrived) workload, even raising the event queue limit by 20x, as proposed in #1579, is not sufficient. These results are derived from commit 78f2af582286b81e6dc9fa9d455ed2b369d933bd in master.
**After.** The event queue length never exceeded 507 for 1000 partitions and 786 for 10000 partitions. In both experiments, the number of events fit comfortably below 10k, the threshold before we start dropping events.
**Baseline.** Without caching, the maximum queue length was 193 for 1000 partitions and 423 for 10000 partitions.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15726141
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,263 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * A common consumption pattern is to access only the blocks that belong to a specific RDD.
+ * For this use case, we should avoid linearly scanning through all the blocks, which could
+ * be expensive if there are thousands of blocks on each block manager. Thus, we need to store
+ * RDD blocks and non-RDD blocks separately. In particular, we store RDD blocks in a map
+ * indexed by RDD IDs, so we can filter out the blocks of interest quickly.
- def memUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ * These collections should only be mutated through the add/update/removeBlock methods.
+ */
+ private val _rddBlocks = new mutable.HashMap[Int, mutable.Map[BlockId, BlockStatus]]
+ private val _nonRddBlocks = new mutable.HashMap[BlockId, BlockStatus]
- def diskUsed = blocks.values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * A map of storage information associated with each RDD.
+ *
+ * The key is the ID of the RDD, and the value is a 4-tuple of the following:
+ * (size in memory, size on disk, size in tachyon, storage level)
+ *
+ * This is updated incrementally on each block added, updated or removed, so as to avoid
+ * linearly scanning through all the blocks within an RDD if we're only interested in a
+ * given RDD's storage information.
+ */
+ private val _rddStorageInfo = new mutable.HashMap[Int, (Long, Long, Long, StorageLevel)]
- def diskUsedByRDD(rddId: Int) =
- rddBlocks.filterKeys(_.rddId == rddId).values.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Instantiate a StorageStatus with the given initial blocks. This essentially makes a copy of
+ * the original blocks map such that the fate of this storage status is not tied to the source.
+ */
+ def this(bmid: BlockManagerId, maxMem: Long, initialBlocks: Map[BlockId, BlockStatus]) {
+ this(bmid, maxMem)
+ initialBlocks.foreach { case (bid, bstatus) => addBlock(bid, bstatus) }
+ }
- def memRemaining: Long = maxMem - memUsed
+ /**
+ * Return the blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * contains, get, and size.
+ */
+ def blocks: Map[BlockId, BlockStatus] = _nonRddBlocks ++ rddBlocks
- def rddBlocks = blocks.collect { case (rdd: RDDBlockId, status) => (rdd, status) }
-}
+ /**
+ * Return the RDD blocks stored in this block manager.
+ *
+ * Note that this is somewhat expensive, as it involves cloning the underlying maps and then
+ * concatenating them together. Much faster alternatives exist for common operations such as
+ * getting the memory, disk, and off-heap memory sizes occupied by this RDD.
+ */
+ def rddBlocks: Map[BlockId, BlockStatus] = _rddBlocks.flatMap { case (_, blocks) => blocks }
-/** Helper methods for storage-related objects. */
-private[spark] object StorageUtils {
+ /** Return the blocks that belong to the given RDD stored in this block manager. */
+ def rddBlocksById(rddId: Int): Map[BlockId, BlockStatus] = {
+ _rddBlocks.get(rddId).getOrElse(Map.empty)
+ }
+
+ /** Add the given block to this storage status. If it already exists, overwrite it. */
+ def addBlock(blockId: BlockId, blockStatus: BlockStatus): Unit = {
+ blockId match {
+ case RDDBlockId(rddId, _) =>
+ // Update the storage info of the RDD, keeping track of any existing status for this block
+ val oldBlockStatus = getBlock(blockId).getOrElse(BlockStatus.empty)
+ val changeInMem = blockStatus.memSize - oldBlockStatus.memSize
--- End diff --
Yeah, the storage level is precisely the reason why I did not do it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50946929
@concretevitamin What do you mean? I am using the default set of listeners attached to the SparkUI.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50955867
QA results for PR 1679:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17755/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by shivaram <gi...@git.apache.org>.
Github user shivaram commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50947160
@andrewor14 Thanks a lot for fixing this. I am testing this out on 200-node EC2 cluster right now and so far it looks good. Its a long running job though, so I'll report back if I notice any problems.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1679#issuecomment-50812922
QA results for PR 1679:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17596/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2316] Avoid O(blocks) operations in lis...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/1679#discussion_r15725810
--- Diff: core/src/main/scala/org/apache/spark/storage/StorageUtils.scala ---
@@ -20,122 +20,263 @@ package org.apache.spark.storage
import scala.collection.Map
import scala.collection.mutable
-import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
import org.apache.spark.annotation.DeveloperApi
/**
* :: DeveloperApi ::
* Storage information for each BlockManager.
+ *
+ * This class assumes BlockId and BlockStatus are immutable, such that the consumers of this
+ * class cannot mutate the source of the information. Accesses are not thread-safe.
*/
@DeveloperApi
-class StorageStatus(
- val blockManagerId: BlockManagerId,
- val maxMem: Long,
- val blocks: mutable.Map[BlockId, BlockStatus] = mutable.Map.empty) {
+class StorageStatus(val blockManagerId: BlockManagerId, val maxMem: Long) {
- def memUsed = blocks.values.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
+ /**
+ * Internal representation of the blocks stored in this block manager.
+ *
+ * A common consumption pattern is to access only the blocks that belong to a specific RDD.
--- End diff --
This is a bit wordy - I think just saying something short like:
```
/* We store rdd-related blocks separately for faster retrieval. */
```
Could convey as much information as this does.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---