You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/05/04 23:05:50 UTC
[GitHub] [spark] HeartSaVioR commented on a change in pull request #28412: [SPARK-31608][CORE][WEBUI] Add a new type of KVStore to make loading UI faster
HeartSaVioR commented on a change in pull request #28412:
URL: https://github.com/apache/spark/pull/28412#discussion_r419768275
##########
File path: core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
##########
@@ -1167,6 +1168,58 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
// At this point the disk data either does not exist or was deleted because it failed to
// load, so the event log needs to be replayed.
+ // TODO: Maybe need to do other check to see if there's enough memory to
+ // use inMemoryStore.
+ if (hybridKVStoreEnabled) {
+ logInfo("Using HybridKVStore as KVStore")
+ var retried = false
+ var store: HybridKVStore = null
+ while(store == null) {
+ val reader = EventLogFileReader(fs, new Path(logDir, attempt.logPath),
+ attempt.lastIndex)
+ val isCompressed = reader.compressionCodec.isDefined
+ logInfo(s"Leasing disk manager space for app $appId / ${attempt.info.attemptId}...")
+ val lease = dm.lease(reader.totalSize, isCompressed)
+ try {
+ val s = new HybridKVStore()
+ val levelDB = KVUtils.open(lease.tmpPath, metadata)
+ s.setLevelDB(levelDB)
+
+ s.startBackgroundThreadToWriteToDB(new HybridKVStore.SwitchingToLevelDBListener {
+ override def onSwitchingToLevelDBSuccess: Unit = {
+ levelDB.close()
+ val newStorePath = lease.commit(appId, attempt.info.attemptId)
+ s.setLevelDB(KVUtils.open(newStorePath, metadata))
+ logInfo(s"Completely switched to use leveldb for app" +
+ s" $appId / ${attempt.info.attemptId}")
+ }
+
+ override def onSwitchingToLevelDBFail(e: Exception): Unit = {
+ logWarning(s"Failed to switch to use LevelDb for app" +
+ s" $appId / ${attempt.info.attemptId}")
+ levelDB.close()
+ throw e
Review comment:
1. Here we should rollback the lease as well.
2. SHS would assume KVStore is loaded properly when the caller method returns. If I understand correctly, throwing exception here doesn't propagate to the caller method, which means HybridKVStore should still serve the content in any way, in-memory store for this case. (Caution of memory usage)
##########
File path: core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
##########
@@ -1167,6 +1168,58 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
// At this point the disk data either does not exist or was deleted because it failed to
// load, so the event log needs to be replayed.
+ // TODO: Maybe need to do other check to see if there's enough memory to
+ // use inMemoryStore.
+ if (hybridKVStoreEnabled) {
+ logInfo("Using HybridKVStore as KVStore")
+ var retried = false
+ var store: HybridKVStore = null
+ while(store == null) {
+ val reader = EventLogFileReader(fs, new Path(logDir, attempt.logPath),
+ attempt.lastIndex)
+ val isCompressed = reader.compressionCodec.isDefined
+ logInfo(s"Leasing disk manager space for app $appId / ${attempt.info.attemptId}...")
+ val lease = dm.lease(reader.totalSize, isCompressed)
+ try {
+ val s = new HybridKVStore()
+ val levelDB = KVUtils.open(lease.tmpPath, metadata)
+ s.setLevelDB(levelDB)
+
+ s.startBackgroundThreadToWriteToDB(new HybridKVStore.SwitchingToLevelDBListener {
+ override def onSwitchingToLevelDBSuccess: Unit = {
+ levelDB.close()
+ val newStorePath = lease.commit(appId, attempt.info.attemptId)
+ s.setLevelDB(KVUtils.open(newStorePath, metadata))
+ logInfo(s"Completely switched to use leveldb for app" +
+ s" $appId / ${attempt.info.attemptId}")
+ }
+
+ override def onSwitchingToLevelDBFail(e: Exception): Unit = {
+ logWarning(s"Failed to switch to use LevelDb for app" +
+ s" $appId / ${attempt.info.attemptId}")
+ levelDB.close()
+ throw e
+ }
+ })
+
+ rebuildAppStore(s, reader, attempt.info.lastUpdated.getTime())
+ s.stopBackgroundThreadAndSwitchToLevelDB()
+ store = s
+ } catch {
+ case _: IOException if !retried =>
+ // compaction may touch the file(s) which app rebuild wants to read
+ // compaction wouldn't run in short interval, so try again...
+ logWarning(s"Exception occurred while rebuilding app $appId - trying again...")
+ lease.rollback()
Review comment:
`lease.rollback()` was enough to throw out intermediate LevelDB KVStore on temporary directory which fails to load at any reason. It doesn't look like the case of Hybrid KVStore.
##########
File path: core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
##########
@@ -1167,6 +1168,58 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
// At this point the disk data either does not exist or was deleted because it failed to
// load, so the event log needs to be replayed.
+ // TODO: Maybe need to do other check to see if there's enough memory to
+ // use inMemoryStore.
+ if (hybridKVStoreEnabled) {
Review comment:
I've feeling that too much HybridKVStore implementation details are exposed here which could be abstracted away if we have proper interface for disk based loading-only KVStore for SHS.
E.g. suppose new interface on top of KVStore receives a Lease on `initialize()`, and exposes `commit()` & `rollback()` to handle the implementation details on each condition.
##########
File path: core/src/main/scala/org/apache/spark/internal/config/History.scala
##########
@@ -195,4 +195,9 @@ private[spark] object History {
.version("3.0.0")
.booleanConf
.createWithDefault(true)
+
+ val HYBRID_KVSTORE_ENABLED = ConfigBuilder("spark.history.store.hybridKVStore.enabled")
+ .version("3.0.1")
Review comment:
1. This needs to have doc to describe the functionality, as well as proper caution on memory usage.
2. "3.1.0" is correct as of now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org