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 all 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>
</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
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);
Copy link
Contributor

@mridulm mridulm Aug 26, 2022 •
edited
Loading

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
4 changes: 2 additions & 2 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 @@ -162,8 +162,8 @@ 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)
Expand Down
4 changes: 3 additions & 1 deletion ...-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,7 +45,7 @@ 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
Expand Down Expand Up @@ -1068,6 +1068,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