Skip to content

[SPARK-38909][BUILD][CORE][YARN][FOLLOWUP] Make some code cleanup related to shuffle state db #37648

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
LuciferYang wants to merge 22 commits into from
Closed

[SPARK-38909][BUILD][CORE][YARN][FOLLOWUP] Make some code cleanup related to shuffle state db #37648

Show file tree
Hide file tree
Changes from 16 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
579956e
fix wuyi's comments
LuciferYang Aug 24, 2022
2caffd1
add @DeveloperApi annotation
LuciferYang Aug 25, 2022
87c42dd
Revert "add @DeveloperApi annotation"
LuciferYang Aug 26, 2022
e3500cf
remove condition
LuciferYang Aug 26, 2022
86681e3
part revert
LuciferYang Aug 26, 2022
7e3befa
Revert "part revert"
LuciferYang Aug 26, 2022
6ceb2d0
fix test
LuciferYang Aug 26, 2022
11b9dad
delete remove from LevelDBIteraot
LuciferYang Aug 26, 2022
8b77b61
make db and dbiterator private
LuciferYang Aug 27, 2022
3e59b7b
Merge branch 'upmaster' into SPARK-38909-followup
LuciferYang Aug 29, 2022
7a22aeb
move test only initDB back to ShuffleTestAccessor
LuciferYang Aug 29, 2022
9449457
Merge branch 'upmaster' into SPARK-38909-followup
LuciferYang Aug 31, 2022
1f3b62b
Revert "move test only initDB back to ShuffleTestAccessor"
LuciferYang Aug 31, 2022
ec370e7
use defualt initDB
LuciferYang Aug 31, 2022
11850af
imports
LuciferYang Aug 31, 2022
024ec63
Merge branch 'upmaster' into SPARK-38909-followup
LuciferYang Sep 2, 2022
c8ac7e4
revert blank line
LuciferYang Sep 2, 2022
33c96a6
Merge branch 'upmaster' into SPARK-38909-followup
LuciferYang Sep 6, 2022
a5df1c4
Revert "use defualt initDB"
LuciferYang Sep 7, 2022
dfe7ce5
clean up improts
LuciferYang Sep 7, 2022
6c91380
clean up improts
LuciferYang Sep 7, 2022
8f47c15
Merge branch 'upmaster' into SPARK-38909-followup
LuciferYang Sep 7, 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
1 change: 0 additions & 1 deletion common/network-common/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,6 @@
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-tags_${scala.binary.version}</artifactId>
<scope>test</scope>
Comment thread
mridulm marked this conversation as resolved.
</dependency>

<!--
Expand Down
3 changes: 3 additions & 0 deletions common/network-common/src/main/java/org/apache/spark/network/shuffledb/DB.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,13 @@

import java.io.Closeable;

import org.apache.spark.annotation.Private;

/**
* The local KV storage used to persist the shuffle state,
* the implementations may include LevelDB, RocksDB, etc.
*/
@Private
public interface DB extends Closeable {
/**
* Set the DB entry for "key" to "value".
Expand Down
3 changes: 3 additions & 0 deletions common/network-common/src/main/java/org/apache/spark/network/shuffledb/DBIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@
import java.util.Iterator;
import java.util.Map;

import org.apache.spark.annotation.Private;

@Private
public interface DBIterator extends Iterator<Map.Entry<byte[], byte[]>>, Closeable {

/**
Expand Down
5 changes: 0 additions & 5 deletions common/network-common/src/main/java/org/apache/spark/network/shuffledb/LevelDBIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,11 +78,6 @@ public void seek(byte[] key) {
it.seek(key);
}

@Override
public void remove() {
throw new UnsupportedOperationException();
}

private Map.Entry<byte[], byte[]> loadNext() {
boolean hasNext = it.hasNext();
if (!hasNext) {
Expand Down
14 changes: 0 additions & 14 deletions common/network-common/src/main/java/org/apache/spark/network/util/DBProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
import java.io.IOException;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;

import org.apache.spark.network.shuffledb.DBBackend;
import org.apache.spark.network.shuffledb.LevelDB;
Expand All @@ -45,17 +44,4 @@ public static DB initDB(
}
return null;
}

@VisibleForTesting
public static DB initDB(DBBackend dbBackend, File file) throws IOException {
if (file != null) {
// TODO: SPARK-38888, add rocksdb implementation.
switch (dbBackend) {
case LEVELDB: return new LevelDB(LevelDBProvider.initLevelDB(file));
default:
throw new IllegalArgumentException("Unsupported DBBackend: " + dbBackend);
}
}
return null;
}
}
9 changes: 0 additions & 9 deletions common/network-common/src/main/java/org/apache/spark/network/util/LevelDBProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import java.io.IOException;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import org.fusesource.leveldbjni.JniDBFactory;
import org.fusesource.leveldbjni.internal.NativeDB;
import org.iq80.leveldb.DB;
Expand Down Expand Up @@ -85,14 +84,6 @@ public static DB initLevelDB(File dbFile, StoreVersion version, ObjectMapper map
return tmpDb;
}

@VisibleForTesting
static DB initLevelDB(File file) throws IOException {
Options options = new Options();
options.createIfMissing(true);
JniDBFactory factory = new JniDBFactory();
return factory.open(file, options);
}

Comment thread
mridulm marked this conversation as resolved.
private static class LevelDBLogger implements org.iq80.leveldb.Logger {
private static final Logger LOG = LoggerFactory.getLogger(LevelDBLogger.class);

Expand Down
13 changes: 5 additions & 8 deletions ...-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,16 +125,13 @@ public ShuffleIndexInformation load(String filePath) throws IOException {
.weigher((Weigher<String, ShuffleIndexInformation>)
(filePath, indexInfo) -> indexInfo.getRetainedMemorySize())
.build(indexCacheLoader);
DBBackend dbBackend = null;
if (registeredExecutorFile != null) {
String dbBackendName =
conf.get(Constants.SHUFFLE_SERVICE_DB_BACKEND, DBBackend.LEVELDB.name());
dbBackend = DBBackend.byName(dbBackendName);
logger.info("Configured {} as {} and actually used value {}",
Constants.SHUFFLE_SERVICE_DB_BACKEND, dbBackendName, dbBackend);
}
String dbBackendName =
conf.get(Constants.SHUFFLE_SERVICE_DB_BACKEND, DBBackend.LEVELDB.name());
DBBackend dbBackend = DBBackend.byName(dbBackendName);
db = DBProvider.initDB(dbBackend, this.registeredExecutorFile, CURRENT_VERSION, mapper);

@mridulm mridulm Aug 26, 2022 •
edited
Loading

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.

Review note: Recovery need not be enabled for node managers - in which case registeredExecutorFile will be null (in addition to tests).

DBProvider.initDB does handle null input though.

So the main change in this file and RemoteBlockPushResolver is moving the log message into if (db != null)

if (db != null) {
logger.info("Use {} as the implementation of {}",
dbBackend, Constants.SHUFFLE_SERVICE_DB_BACKEND);
executors = reloadRegisteredExecutors(db);
} else {
executors = Maps.newConcurrentMap();
Expand Down
13 changes: 5 additions & 8 deletions ...twork-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,16 +156,13 @@ public ShuffleIndexInformation load(String filePath) throws IOException {
(filePath, indexInfo) -> indexInfo.getRetainedMemorySize())
.build(indexCacheLoader);
this.recoveryFile = recoveryFile;
DBBackend dbBackend = null;
if (recoveryFile != null) {
String dbBackendName =
conf.get(Constants.SHUFFLE_SERVICE_DB_BACKEND, DBBackend.LEVELDB.name());
dbBackend = DBBackend.byName(dbBackendName);
logger.info("Configured {} as {} and actually used value {}",
Constants.SHUFFLE_SERVICE_DB_BACKEND, dbBackendName, dbBackend);
}
String dbBackendName =
conf.get(Constants.SHUFFLE_SERVICE_DB_BACKEND, DBBackend.LEVELDB.name());
DBBackend dbBackend = DBBackend.byName(dbBackendName);
db = DBProvider.initDB(dbBackend, this.recoveryFile, CURRENT_VERSION, mapper);
if (db != null) {
logger.info("Use {} as the implementation of {}",
dbBackend, Constants.SHUFFLE_SERVICE_DB_BACKEND);
reloadAndCleanUpAppShuffleInfo(db);
}
}
Expand Down
4 changes: 2 additions & 2 deletions common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,8 +241,8 @@ protected void serviceInit(Configuration externalConf) throws Exception {
String dbBackendName = _conf.get(Constants.SHUFFLE_SERVICE_DB_BACKEND,
DBBackend.LEVELDB.name());
dbBackend = DBBackend.byName(dbBackendName);
logger.info("Configured {} as {} and actually used value {}",
Constants.SHUFFLE_SERVICE_DB_BACKEND, dbBackendName, dbBackend);
logger.info("Use {} as the implementation of {}",
dbBackend, Constants.SHUFFLE_SERVICE_DB_BACKEND);
}

try {
Expand Down
8 changes: 4 additions & 4 deletions core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,11 +81,11 @@ class ExternalShuffleService(sparkConf: SparkConf, securityManager: SecurityMana
protected def newShuffleBlockHandler(conf: TransportConf): ExternalBlockHandler = {
if (sparkConf.get(config.SHUFFLE_SERVICE_DB_ENABLED) && enabled) {
val shuffleDBName = sparkConf.get(config.SHUFFLE_SERVICE_DB_BACKEND)
val dBBackend = DBBackend.byName(shuffleDBName)
logInfo(s"Configured ${config.SHUFFLE_SERVICE_DB_BACKEND.key} as $shuffleDBName " +
s"and actually used value ${dBBackend.name()} ")
val dbBackend = DBBackend.byName(shuffleDBName)
logInfo(s"Use ${dbBackend.name()} as the implementation of " +
s"${config.SHUFFLE_SERVICE_DB_BACKEND.key}")
new ExternalBlockHandler(conf,
findRegisteredExecutorsDBFile(dBBackend.fileName(registeredExecutorsDB)))
findRegisteredExecutorsDBFile(dbBackend.fileName(registeredExecutorsDB)))
} else {
new ExternalBlockHandler(conf, null)
}
Expand Down
11 changes: 6 additions & 5 deletions ...nagers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnShuffleIntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ package org.apache.spark.deploy.yarn
import java.io.File
import java.nio.charset.StandardCharsets

import com.fasterxml.jackson.databind.ObjectMapper
import com.google.common.io.Files
import org.apache.commons.io.FileUtils
import org.apache.hadoop.yarn.conf.YarnConfiguration
Expand All @@ -32,7 +33,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Network._
import org.apache.spark.network.shuffle.ShuffleTestAccessor
import org.apache.spark.network.shuffledb.DBBackend
import org.apache.spark.network.shuffledb.{DBBackend, StoreVersion}
import org.apache.spark.network.yarn.{YarnShuffleService, YarnTestAccessor}
import org.apache.spark.tags.ExtendedYarnTest

Expand Down Expand Up @@ -162,11 +163,11 @@ private object YarnExternalShuffleDriver extends Logging with Matchers {
if (registeredExecFile != null && execStateCopy != null) {
val dbBackendName = conf.get(SHUFFLE_SERVICE_DB_BACKEND.key)
val dbBackend = DBBackend.byName(dbBackendName)
logWarning(s"Configured ${SHUFFLE_SERVICE_DB_BACKEND.key} as $dbBackendName " +
s"and actually used value ${dbBackend.name()}")
logWarning(s"Use ${dbBackend.name()} as the implementation of " +
s"${SHUFFLE_SERVICE_DB_BACKEND.key}")
FileUtils.copyDirectory(registeredExecFile, execStateCopy)
assert(!ShuffleTestAccessor
.reloadRegisteredExecutors(dbBackend, execStateCopy).isEmpty)
assert(!ShuffleTestAccessor.reloadRegisteredExecutors(
dbBackend, execStateCopy, new StoreVersion(1, 0), new ObjectMapper()).isEmpty)
}
} finally {
sc.stop()
Expand Down
13 changes: 8 additions & 5 deletions ...e-managers/yarn/src/test/scala/org/apache/spark/network/shuffle/ShuffleTestAccessor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,13 @@ import java.nio.channels.FileChannel
import java.util.List
import java.util.concurrent.ConcurrentMap

import com.fasterxml.jackson.databind.ObjectMapper
import org.apache.hadoop.yarn.api.records.ApplicationId

import org.apache.spark.network.shuffle.ExternalShuffleBlockResolver.AppExecId
import org.apache.spark.network.shuffle.RemoteBlockPushResolver._
import org.apache.spark.network.shuffle.protocol.{ExecutorShuffleInfo, FinalizeShuffleMerge}
import org.apache.spark.network.shuffledb.DB
import org.apache.spark.network.shuffledb.DBBackend
import org.apache.spark.network.shuffledb.{DB, DBBackend, StoreVersion}
import org.apache.spark.network.util.{DBProvider, TransportConf}

/**
Expand Down Expand Up @@ -208,9 +208,12 @@ object ShuffleTestAccessor {
}

def reloadRegisteredExecutors(
dbBackend: DBBackend,
file: File): ConcurrentMap[ExternalShuffleBlockResolver.AppExecId, ExecutorShuffleInfo] = {
val db = DBProvider.initDB(dbBackend, file)
dbBackend: DBBackend,
file: File,
version: StoreVersion,
mapper: ObjectMapper)
: ConcurrentMap[ExternalShuffleBlockResolver.AppExecId, ExecutorShuffleInfo] = {
val db = DBProvider.initDB(dbBackend, file, version, mapper)

@mridulm mridulm Sep 6, 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.

This is a behavior change for test, right (this ?)
Essentially, YarnShuffleIntegrationSuite would expect it to fail with an exception and not do error handling (which initDB does).

@LuciferYang LuciferYang Sep 6, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

spark/resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnShuffleIntegrationSuite.scala

Lines 155 to 170 in 19b1780

try {
val data = sc.parallelize(0 until 100, 10).map { x => (x % 10) -> x }.reduceByKey{ _ + _ }.
collect().toSet
sc.listenerBus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS)
data should be ((0 until 10).map{x => x -> (x * 10 + 450)}.toSet)
result = "success"
// only one process can open a leveldb file at a time, so we copy the files
if (registeredExecFile != null && execStateCopy != null) {
val dbBackendName = conf.get(SHUFFLE_SERVICE_DB_BACKEND.key)
val dbBackend = DBBackend.byName(dbBackendName)
logWarning(s"Configured ${SHUFFLE_SERVICE_DB_BACKEND.key} as $dbBackendName " +
s"and actually used value ${dbBackend.name()}")
FileUtils.copyDirectory(registeredExecFile, execStateCopy)
assert(!ShuffleTestAccessor
.reloadRegisteredExecutors(dbBackend, execStateCopy).isEmpty)
}

Do you mean the behavior change is execStateCopy exists but open failed? The original method will throw an exception directly, and the new method will return a new instance? It seems that the case just to test whether an existing execStateCopy can be loaded again?

@LuciferYang LuciferYang Sep 6, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

If the original case does not test the execStateCopy exists but open failed scenario, it seems that there is no change.

For the initDB(dbBackend, file) method:

spark/common/network-common/src/main/java/org/apache/spark/network/util/LevelDBProvider.java

Lines 88 to 94 in 19b1780

@VisibleForTesting
static DB initLevelDB(File file) throws IOException {
Options options = new Options();
options.createIfMissing(true);
JniDBFactory factory = new JniDBFactory();
return factory.open(file, options);
}

  • if execStateCopy exists, it will re-open it and load data from execStateCopy
  • else if execStateCopy not exists, it will open a new db and return empty data due to options.createIfMissing(true), then assert failed

For the initDB(dbBackend, dbFile, version, mapper) method:

  • if execStateCopy exists, it will re-open it and load data from execStateCopy
  • else if execStateCopy not exists, it will open a new db and return empty data, then assert failed

For execStateCopy exists but open failed scenario, use initDB(dbBackend, file) will throw an unhandled exception, use initDB(dbBackend, dbFile, version, mapper) will assert failed. Is this unacceptable?

@LuciferYang LuciferYang Sep 6, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@mridulm any suggestions?

@mridulm mridulm Sep 6, 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.

@squito or @tgravescs should have more context about this, and comment better - looking at the test, I would expect this test code path to additionally test if the DB is missing/invalid/etc and fail in case there is any issues (due to lack of error handling/fallback in LevelDBProvider.initLevelDB) - which changes with in PR, that no longer happens; right ?

@LuciferYang LuciferYang Sep 7, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

done

@LuciferYang LuciferYang Sep 7, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@mridulm add SPARK-40364 to tracking this

@LuciferYang LuciferYang Sep 7, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I will give a separately pr to continue to explore the feasibility about this change

@tgravescs tgravescs Sep 7, 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.

sorry didn't get a chance to look before your filed the other issue, happy to look if you make the change under the other issue

@LuciferYang LuciferYang Sep 7, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Thanks @tgravescs , I will give another pr later ~

val result = ExternalShuffleBlockResolver.reloadRegisteredExecutors(db)
db.close()
result
Expand Down
5 changes: 3 additions & 2 deletions ...-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,15 +45,14 @@ import org.apache.spark.SecurityManager
import org.apache.spark.SparkFunSuite
import org.apache.spark.internal.config._
import org.apache.spark.network.server.BlockPushNonFatalFailure
import org.apache.spark.network.shuffle.{MergedShuffleFileManager, NoOpMergedShuffleFileManager, RemoteBlockPushResolver, ShuffleTestAccessor}
import org.apache.spark.network.shuffle.{Constants, MergedShuffleFileManager, NoOpMergedShuffleFileManager, RemoteBlockPushResolver, ShuffleTestAccessor}
import org.apache.spark.network.shuffle.RemoteBlockPushResolver._
import org.apache.spark.network.shuffle.protocol.ExecutorShuffleInfo
import org.apache.spark.network.shuffledb.DBBackend
import org.apache.spark.network.util.TransportConf
import org.apache.spark.network.yarn.util.HadoopConfigProvider
import org.apache.spark.tags.ExtendedLevelDBTest
import org.apache.spark.util.Utils

@Ngone51 Ngone51 Sep 2, 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.

revert this?

@LuciferYang LuciferYang Sep 2, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

done

abstract class YarnShuffleServiceSuite extends SparkFunSuite with Matchers {

private[yarn] var yarnConfig: YarnConfiguration = null
Expand Down Expand Up @@ -1068,6 +1067,8 @@ abstract class YarnShuffleServiceSuite extends SparkFunSuite with Matchers {

test("create remote block push resolver instance") {
val mockConf = mock(classOf[TransportConf])
when(mockConf.get(Constants.SHUFFLE_SERVICE_DB_BACKEND, DBBackend.LEVELDB.name()))
.thenReturn(shuffleDBBackend().name())
when(mockConf.mergedShuffleFileManagerImpl).thenReturn(
"org.apache.spark.network.shuffle.RemoteBlockPushResolver")
val mergeMgr = YarnShuffleService.newMergedShuffleFileManagerInstance(mockConf, null)
Expand Down