Skip to content

[SPARK-41054][UI][CORE] Support RocksDB as KVStore in live UI #38567

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gengliangwang wants to merge 5 commits into from
Closed

[SPARK-41054][UI][CORE] Support RocksDB as KVStore in live UI #38567

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f3aa7d2
save for now
gengliangwang Nov 7, 2022
2d98b2f
support using rocksdb in live UI
gengliangwang Nov 8, 2022
09f3321
remove comment about replication
gengliangwang Nov 21, 2022
6732462
add doc
gengliangwang Nov 22, 2022
ac2b098
remove one empty line
gengliangwang Nov 22, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 6 additions & 35 deletions core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ package org.apache.spark.deploy.history

import java.io.{File, FileNotFoundException, IOException}
import java.lang.{Long => JLong}
import java.nio.file.Files
import java.util.{Date, NoSuchElementException, ServiceLoader}
import java.util.concurrent.{ConcurrentHashMap, ExecutorService, TimeUnit}
import java.util.zip.ZipOutputStream
Expand All @@ -36,15 +35,12 @@ import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
import org.apache.hadoop.hdfs.DistributedFileSystem
import org.apache.hadoop.hdfs.protocol.HdfsConstants
import org.apache.hadoop.security.AccessControlException
import org.fusesource.leveldbjni.internal.NativeDB
import org.rocksdb.RocksDBException

import org.apache.spark.{SecurityManager, SparkConf, SparkException}
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.Logging
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.History._
import org.apache.spark.internal.config.History.HybridStoreDiskBackend._
import org.apache.spark.internal.config.Status._
import org.apache.spark.internal.config.Tests.IS_TESTING
import org.apache.spark.internal.config.UI._
Expand Down Expand Up @@ -136,34 +132,9 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
HybridStoreDiskBackend.withName(conf.get(History.HYBRID_STORE_DISK_BACKEND))

// Visible for testing.
private[history] val listing: KVStore = storePath.map { path =>
val dir = hybridStoreDiskBackend match {
case LEVELDB => "listing.ldb"
case ROCKSDB => "listing.rdb"
}
val dbPath = Files.createDirectories(new File(path, dir).toPath()).toFile()
Utils.chmod700(dbPath)

val metadata = FsHistoryProviderMetadata(CURRENT_LISTING_VERSION,
AppStatusStore.CURRENT_VERSION, logDir)

try {
open(dbPath, metadata, conf)
} catch {
// If there's an error, remove the listing database and any existing UI database
// from the store directory, since it's extremely likely that they'll all contain
// incompatible information.
case _: UnsupportedStoreVersionException | _: MetadataMismatchException =>
logInfo("Detected incompatible DB versions, deleting...")
path.listFiles().foreach(Utils.deleteRecursively)
open(dbPath, metadata, conf)
case dbExc @ (_: NativeDB.DBException | _: RocksDBException) =>
// Get rid of the corrupted data and re-create it.
logWarning(s"Failed to load disk store $dbPath :", dbExc)
Utils.deleteRecursively(dbPath)
open(dbPath, metadata, conf)
}
}.getOrElse(new InMemoryStore())
private[history] val listing: KVStore = {
KVUtils.createKVStore(storePath, hybridStoreDiskBackend, conf)
}

private val diskManager = storePath.map { path =>
new HistoryServerDiskManager(conf, path, listing, clock)
Expand Down Expand Up @@ -1444,7 +1415,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
}
}

private[history] object FsHistoryProvider {
private[spark] object FsHistoryProvider {

private val APPL_START_EVENT_PREFIX = "{\"Event\":\"SparkListenerApplicationStart\""

Expand All @@ -1459,10 +1430,10 @@ private[history] object FsHistoryProvider {
* db, if the version does not match this value, the FsHistoryProvider will throw away
* all data and re-generate the listing data from the event logs.
*/
private[history] val CURRENT_LISTING_VERSION = 1L
val CURRENT_LISTING_VERSION = 1L
}

private[history] case class FsHistoryProviderMetadata(
private[spark] case class FsHistoryProviderMetadata(
version: Long,
uiVersion: Long,
logDir: String)
Expand Down
7 changes: 7 additions & 0 deletions core/src/main/scala/org/apache/spark/internal/config/Status.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,4 +70,11 @@ private[spark] object Status {
.version("3.0.0")
.booleanConf
.createWithDefault(false)

val LIVE_UI_LOCAL_STORE_DIR = ConfigBuilder("spark.ui.store.path")
.doc("Local directory where to cache application information for live UI. By default this is " +
"not set, meaning all application information will be kept in memory.")
Comment thread
tgravescs marked this conversation as resolved.
.version("3.4.0")
.stringConf
.createOptional
}
14 changes: 12 additions & 2 deletions core/src/main/scala/org/apache/spark/status/AppStatusStore.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,17 +17,20 @@

package org.apache.spark.status

import java.io.File
import java.util.{List => JList}

import scala.collection.JavaConverters._
import scala.collection.mutable.HashMap

import org.apache.spark.{JobExecutionStatus, SparkConf, SparkContext}
import org.apache.spark.internal.config.History.HybridStoreDiskBackend
import org.apache.spark.internal.config.Status.LIVE_UI_LOCAL_STORE_DIR
import org.apache.spark.status.api.v1
import org.apache.spark.storage.FallbackStorage.FALLBACK_BLOCK_MANAGER_ID
import org.apache.spark.ui.scope._
import org.apache.spark.util.Utils
import org.apache.spark.util.kvstore.{InMemoryStore, KVStore}
import org.apache.spark.util.kvstore.KVStore

/**
* A wrapper around a KVStore that provides methods for accessing the API data stored within.
Expand Down Expand Up @@ -769,7 +772,14 @@ private[spark] object AppStatusStore {
def createLiveStore(
conf: SparkConf,
appStatusSource: Option[AppStatusSource] = None): AppStatusStore = {
val store = new ElementTrackingStore(new InMemoryStore(), conf)
val storePath = conf.get(LIVE_UI_LOCAL_STORE_DIR).map(new File(_))
// For the disk-based KV store of live UI, let's simply make it ROCKSDB only for now,
// instead of supporting both LevelDB and RocksDB. RocksDB is built based on LevelDB with

@cloud-fan cloud-fan Nov 23, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not related to this PR, I'm wondering why did we provide these two choices in the first place. RocksDB only seems sufficient.

@gengliangwang gengliangwang Nov 23, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://issues.apache.org/jira/browse/SPARK-18085?focusedCommentId=16088758&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-16088758

There were legal concerns.

@dongjoon-hyun dongjoon-hyun Nov 23, 2022 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

According to the context, I thought Reynold mentioned that the legal issue was resolved. Do you mean that Apache Spark still have legal issue with RocksDB, @gengliangwang ?

Screenshot 2022-11-23 at 12 28 24 PM

@gengliangwang gengliangwang Nov 23, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@dongjoon-hyun I was explaining why @vanzin chose LevelDB. Maybe I was wrong about his reason.

Do you mean that Apache Spark still have legal issue with RocksDB

I don't think so. If so, I wouldn't choose it as the only disk backend of live UI.

@dongjoon-hyun dongjoon-hyun Nov 24, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. Thank you for confirming that.

// improvements on writes and reads. Furthermore, we can reuse the RocksDBFileManager in
// streaming for replicating the local RocksDB file to DFS. The replication in DFS can be
// used for Spark history server.
Comment thread
gengliangwang marked this conversation as resolved.
Outdated
val kvStore = KVUtils.createKVStore(storePath, HybridStoreDiskBackend.ROCKSDB, conf)
val store = new ElementTrackingStore(kvStore, conf)
val listener = new AppStatusListener(store, conf, true, appStatusSource)
new AppStatusStore(store, listener = Some(listener))
}
Expand Down
51 changes: 49 additions & 2 deletions core/src/main/scala/org/apache/spark/status/KVUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,16 +18,21 @@
package org.apache.spark.status

import java.io.File
import java.nio.file.Files

import scala.annotation.meta.getter
import scala.collection.JavaConverters._
import scala.reflect.{classTag, ClassTag}

import com.fasterxml.jackson.annotation.JsonInclude
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.fusesource.leveldbjni.internal.NativeDB
import org.rocksdb.RocksDBException

import org.apache.spark.SparkConf
import org.apache.spark.deploy.history.{FsHistoryProvider, FsHistoryProviderMetadata}
import org.apache.spark.internal.Logging
import org.apache.spark.internal.config.History
import org.apache.spark.internal.config.History.HYBRID_STORE_DISK_BACKEND
import org.apache.spark.internal.config.History.HybridStoreDiskBackend
import org.apache.spark.internal.config.History.HybridStoreDiskBackend._
Expand Down Expand Up @@ -62,10 +67,14 @@ private[spark] object KVUtils extends Logging {
* the store's metadata.
* @param conf SparkConf use to get `HYBRID_STORE_DISK_BACKEND`
*/
def open[M: ClassTag](path: File, metadata: M, conf: SparkConf): KVStore = {
def open[M: ClassTag](
path: File,
metadata: M,
conf: SparkConf,
diskBackend: Option[HybridStoreDiskBackend.Value] = None): KVStore = {
require(metadata != null, "Metadata is required.")

val db = backend(conf) match {
val db = diskBackend.getOrElse(backend(conf)) match {
case LEVELDB => new LevelDB(path, new KVStoreScalaSerializer())
case ROCKSDB => new RocksDB(path, new KVStoreScalaSerializer())
}
Expand All @@ -80,6 +89,44 @@ private[spark] object KVUtils extends Logging {
db
}

def createKVStore(
storePath: Option[File],
diskBackend: HybridStoreDiskBackend.Value,
conf: SparkConf): KVStore = {
storePath.map { path =>
val dir = diskBackend match {
case LEVELDB => "listing.ldb"
case ROCKSDB => "listing.rdb"
}

val dbPath = Files.createDirectories(new File(path, dir).toPath()).toFile()
Utils.chmod700(dbPath)


Comment thread
tgravescs marked this conversation as resolved.
Outdated
val metadata = FsHistoryProviderMetadata(
FsHistoryProvider.CURRENT_LISTING_VERSION,
AppStatusStore.CURRENT_VERSION,
conf.get(History.HISTORY_LOG_DIR))

try {
open(dbPath, metadata, conf, Some(diskBackend))
} catch {

@LuciferYang LuciferYang Nov 17, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For Live UI, are these two exception scenarios possible?

@gengliangwang gengliangwang Nov 17, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No. The LevelDB exception is not possible either. There are only two error-handling branches, and seems too much to distinguish whether the caller is for live UI or SHS.

@LuciferYang LuciferYang Nov 21, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If unexpected data cleaning doesn't occur in the live UI scenario, I think it is OK

@tgravescs tgravescs Nov 21, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If something goes wrong and the kv store can't be created for live ui then the entire application fails?

@gengliangwang gengliangwang Nov 22, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@tgravescs If there is exception from RocksDB, it will delete the target directory and open the KV store again. Otherwise if the error is not from RocksDB, it will fall back to the in-memory store.
It follows the same logic in SHS here.

// If there's an error, remove the listing database and any existing UI database
// from the store directory, since it's extremely likely that they'll all contain
// incompatible information.
case _: UnsupportedStoreVersionException | _: MetadataMismatchException =>
logInfo("Detected incompatible DB versions, deleting...")
path.listFiles().foreach(Utils.deleteRecursively)
open(dbPath, metadata, conf, Some(diskBackend))
case dbExc @ (_: NativeDB.DBException | _: RocksDBException) =>
// Get rid of the corrupted data and re-create it.
logWarning(s"Failed to load disk store $dbPath :", dbExc)
Utils.deleteRecursively(dbPath)
open(dbPath, metadata, conf, Some(diskBackend))
}
}.getOrElse(new InMemoryStore())
}

/** Turns a KVStoreView into a Scala sequence, applying a filter. */
def viewToSeq[T](
view: KVStoreView[T],
Expand Down
19 changes: 16 additions & 3 deletions core/src/test/scala/org/apache/spark/status/AppStatusStoreSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import scala.util.Random
import org.apache.spark.{SparkConf, SparkFunSuite}
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.internal.config.History.{HYBRID_STORE_DISK_BACKEND, HybridStoreDiskBackend}
import org.apache.spark.internal.config.Status.LIVE_ENTITY_UPDATE_PERIOD
import org.apache.spark.internal.config.Status.{LIVE_ENTITY_UPDATE_PERIOD, LIVE_UI_LOCAL_STORE_DIR}
import org.apache.spark.resource.ResourceProfile
import org.apache.spark.scheduler.{SparkListenerStageSubmitted, SparkListenerTaskStart, StageInfo, TaskInfo, TaskLocality}
import org.apache.spark.status.api.v1.SpeculationStageSummary
Expand Down Expand Up @@ -88,7 +88,19 @@ class AppStatusStoreSuite extends SparkFunSuite {
live: Boolean): AppStatusStore = {
val conf = new SparkConf()
if (live) {
return AppStatusStore.createLiveStore(conf)
if (disk) {
val testDir = Utils.createTempDir()
conf.set(LIVE_UI_LOCAL_STORE_DIR, testDir.getCanonicalPath)
}
val liveStore = AppStatusStore.createLiveStore(conf)
if (disk) {
val rocksDBCreated = liveStore.store match {
case e: ElementTrackingStore => !e.usingInMemoryStore
case _ => false
}
assert(rocksDBCreated)
}
return liveStore
}

val store: KVStore = if (disk) {
Expand All @@ -106,7 +118,8 @@ class AppStatusStoreSuite extends SparkFunSuite {
val baseCases = Seq(
"disk rocksdb" -> createAppStore(disk = true, HybridStoreDiskBackend.ROCKSDB, live = false),
"in memory" -> createAppStore(disk = false, live = false),
"in memory live" -> createAppStore(disk = false, live = true)
"in memory live" -> createAppStore(disk = false, live = true),
"rocksdb live" -> createAppStore(disk = true, HybridStoreDiskBackend.ROCKSDB, live = true)
)
if (Utils.isMacOnAppleSilicon) {
baseCases
Expand Down