From 6a8abe29ef3369b387d9bc2ee3459a6611246ab1 Mon Sep 17 00:00:00 2001
From: zhoukang <zhoukang199191@gmail.com>
Date: Wed, 28 Feb 2018 23:16:29 +0800
Subject: [PATCH] [SPARK-23508][CORE] Fix BlockmanagerId in case
blockManagerIdCache cause oom
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
… cause oom
## What changes were proposed in this pull request?
blockManagerIdCache in BlockManagerId will not remove old values which may cause oom
`val blockManagerIdCache = new ConcurrentHashMap[BlockManagerId, BlockManagerId]()`
Since whenever we apply a new BlockManagerId, it will put into this map.
This patch will use guava cahce for blockManagerIdCache instead.
A heap dump show in [SPARK-23508](https://issues.apache.org/jira/browse/SPARK-23508)
## How was this patch tested?
Exist tests.
Author: zhoukang <zhoukang199191@gmail.com>
Closes #20667 from caneGuy/zhoukang/fix-history.
---
.../org/apache/spark/storage/BlockManagerId.scala | 14 +++++++++++---
1 file changed, 11 insertions(+), 3 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerId.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerId.scala
index 2c3da0ee85..d4a59c33b9 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerId.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerId.scala
@@ -18,7 +18,8 @@
package org.apache.spark.storage
import java.io.{Externalizable, IOException, ObjectInput, ObjectOutput}
-import java.util.concurrent.ConcurrentHashMap
+
+import com.google.common.cache.{CacheBuilder, CacheLoader}
import org.apache.spark.SparkContext
import org.apache.spark.annotation.DeveloperApi
@@ -132,10 +133,17 @@ private[spark] object BlockManagerId {
getCachedBlockManagerId(obj)
}
- val blockManagerIdCache = new ConcurrentHashMap[BlockManagerId, BlockManagerId]()
+ /**
+ * The max cache size is hardcoded to 10000, since the size of a BlockManagerId
+ * object is about 48B, the total memory cost should be below 1MB which is feasible.
+ */
+ val blockManagerIdCache = CacheBuilder.newBuilder()
+ .maximumSize(10000)
+ .build(new CacheLoader[BlockManagerId, BlockManagerId]() {
+ override def load(id: BlockManagerId) = id
+ })
def getCachedBlockManagerId(id: BlockManagerId): BlockManagerId = {
- blockManagerIdCache.putIfAbsent(id, id)
blockManagerIdCache.get(id)
}
}
--
GitLab