You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by an...@apache.org on 2016/04/05 21:32:15 UTC
spark git commit: [SPARK-14243][CORE][BACKPORT-1.6] update task
metrics when removing blocks
Repository: spark
Updated Branches:
refs/heads/branch-1.6 285cb9c66 -> cfe9f02ca
[SPARK-14243][CORE][BACKPORT-1.6] update task metrics when removing blocks
## What changes were proposed in this pull request?
This patch try to update the `updatedBlockStatuses ` when removing blocks, making sure `BlockManager` correctly updates `updatedBlockStatuses`
## How was this patch tested?
test("updated block statuses") in BlockManagerSuite.scala
Author: jeanlyn <je...@gmail.com>
Closes #12150 from jeanlyn/updataBlock1.6.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/cfe9f02c
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/cfe9f02c
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/cfe9f02c
Branch: refs/heads/branch-1.6
Commit: cfe9f02ca81b2c47f8425b3a7c278149caba58d2
Parents: 285cb9c
Author: jeanlyn <je...@gmail.com>
Authored: Tue Apr 5 12:32:03 2016 -0700
Committer: Andrew Or <an...@databricks.com>
Committed: Tue Apr 5 12:32:03 2016 -0700
----------------------------------------------------------------------
.../org/apache/spark/storage/BlockManager.scala | 7 ++++++-
.../spark/storage/BlockManagerSuite.scala | 21 ++++++++++++++++++++
2 files changed, 27 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/cfe9f02c/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
index 2cc2fd9..538272d 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -1120,10 +1120,15 @@ private[spark] class BlockManager(
"the disk, memory, or external block store")
}
blockInfo.remove(blockId)
+ val status = getCurrentBlockStatus(blockId, info)
if (tellMaster && info.tellMaster) {
- val status = getCurrentBlockStatus(blockId, info)
reportBlockStatus(blockId, info, status)
}
+ Option(TaskContext.get()).foreach { tc =>
+ val metrics = tc.taskMetrics()
+ val lastUpdatedBlocks = metrics.updatedBlocks.getOrElse(Seq[(BlockId, BlockStatus)]())
+ metrics.updatedBlocks = Some(lastUpdatedBlocks ++ Seq((blockId, status)))
+ }
}
} else {
// The block has already been removed; do nothing.
http://git-wip-us.apache.org/repos/asf/spark/blob/cfe9f02c/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
index 53991d8..c00591f 100644
--- a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
@@ -895,6 +895,17 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with BeforeAndAfterE
val list = List.fill(2)(new Array[Byte](2000))
val bigList = List.fill(8)(new Array[Byte](2000))
+ def getUpdatedBlocks(task: => Unit): Seq[(BlockId, BlockStatus)] = {
+ val context = TaskContext.empty()
+ try {
+ TaskContext.setTaskContext(context)
+ task
+ } finally {
+ TaskContext.unset()
+ }
+ context.taskMetrics.updatedBlocks.getOrElse(Seq[(BlockId, BlockStatus)]())
+ }
+
// 1 updated block (i.e. list1)
val updatedBlocks1 =
store.putIterator("list1", list.iterator, StorageLevel.MEMORY_ONLY, tellMaster = true)
@@ -954,6 +965,16 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with BeforeAndAfterE
assert(!store.diskStore.contains("list3"), "list3 was in disk store")
assert(!store.diskStore.contains("list4"), "list4 was in disk store")
assert(!store.diskStore.contains("list5"), "list5 was in disk store")
+
+ // remove block - list2 should be removed from disk
+ val updatedBlocks6 = getUpdatedBlocks {
+ store.removeBlock(
+ "list2", tellMaster = true)
+ }
+ assert(updatedBlocks6.size === 1)
+ assert(updatedBlocks6.head._1 === TestBlockId("list2"))
+ assert(updatedBlocks6.head._2.storageLevel == StorageLevel.NONE)
+ assert(!store.diskStore.contains("list2"), "list2 was in disk store")
}
test("query block statuses") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org