Skip to content

[SPARK-17321][YARN] Avoid writing shuffle metadata to disk if NM recovery is disabled #19032

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
jerryshao wants to merge 4 commits into from
Closed

[SPARK-17321][YARN] Avoid writing shuffle metadata to disk if NM recovery is disabled #19032

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5abbe75
Avoid writing shuffle metadata to disk if NM recovery is disabled
jerryshao Aug 24, 2017
185d3dd
Address the comments
jerryshao Aug 28, 2017
9fe7e99
Change the comment
jerryshao Aug 28, 2017
ebe0a24
Update the comments
jerryshao Aug 31, 2017
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
117 changes: 60 additions & 57 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 @@ -29,12 +29,14 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.api.*;
import org.apache.spark.network.util.LevelDBProvider;
import org.iq80.leveldb.DB;
Expand Down Expand Up @@ -73,6 +75,8 @@
public class YarnShuffleService extends AuxiliaryService {
private static final Logger logger = LoggerFactory.getLogger(YarnShuffleService.class);

private static final boolean DEFAULT_NM_RECOVERY_ENABLED = false;

@vanzin vanzin Aug 25, 2017

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.

Isn't this in YarnConfiguration?

@jerryshao jerryshao Aug 28, 2017

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.

Let me check the yarn code.


// Port on which the shuffle server listens for fetch requests
private static final String SPARK_SHUFFLE_SERVICE_PORT_KEY = "spark.shuffle.service.port";
private static final int DEFAULT_SPARK_SHUFFLE_SERVICE_PORT = 7337;
Expand Down Expand Up @@ -153,14 +157,18 @@ protected void serviceInit(Configuration conf) throws Exception {
_conf = conf;

boolean stopOnFailure = conf.getBoolean(STOP_ON_FAILURE_KEY, DEFAULT_STOP_ON_FAILURE);
boolean recoveryEnabled = conf.getBoolean(YarnConfiguration.NM_RECOVERY_ENABLED,
DEFAULT_NM_RECOVERY_ENABLED);

try {
// In case this NM was killed while there were running spark applications, we need to restore
// lost state for the existing executors. We look for an existing file in the NM's local dirs.
// If we don't find one, then we choose a file to use to save the state next time. Even if
// an application was stopped while the NM was down, we expect yarn to call stopApplication()
// when it comes back
registeredExecutorFile = initRecoveryDb(RECOVERY_FILE_NAME);
if (recoveryEnabled) {
registeredExecutorFile = initRecoveryDb(RECOVERY_FILE_NAME);
}

TransportConf transportConf = new TransportConf("shuffle", new HadoopConfigProvider(conf));
blockHandler = new ExternalShuffleBlockHandler(transportConf, registeredExecutorFile);
Expand All @@ -170,7 +178,7 @@ protected void serviceInit(Configuration conf) throws Exception {
List<TransportServerBootstrap> bootstraps = Lists.newArrayList();
boolean authEnabled = conf.getBoolean(SPARK_AUTHENTICATE_KEY, DEFAULT_SPARK_AUTHENTICATE);
if (authEnabled) {
createSecretManager();
createSecretManager(recoveryEnabled);

@vanzin vanzin Aug 25, 2017

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.

I think at this point it would be cleaner to do:

secretManager = new ShuffleSecretManager();
if (recoveryEnabled) {
  loadSecretsFromDb();
}

bootstraps.add(new AuthServerBootstrap(transportConf, secretManager));
}

Expand All @@ -194,30 +202,33 @@ protected void serviceInit(Configuration conf) throws Exception {
}
}

private void createSecretManager() throws IOException {
private void createSecretManager(boolean recoveryEnabled) throws IOException {
secretManager = new ShuffleSecretManager();
secretsFile = initRecoveryDb(SECRETS_RECOVERY_FILE_NAME);

// Make sure this is protected in case its not in the NM recovery dir
FileSystem fs = FileSystem.getLocal(_conf);
fs.mkdirs(new Path(secretsFile.getPath()), new FsPermission((short)0700));

db = LevelDBProvider.initLevelDB(secretsFile, CURRENT_VERSION, mapper);
logger.info("Recovery location is: " + secretsFile.getPath());
if (db != null) {
logger.info("Going to reload spark shuffle data");
DBIterator itr = db.iterator();
itr.seek(APP_CREDS_KEY_PREFIX.getBytes(StandardCharsets.UTF_8));
while (itr.hasNext()) {
Map.Entry<byte[], byte[]> e = itr.next();
String key = new String(e.getKey(), StandardCharsets.UTF_8);
if (!key.startsWith(APP_CREDS_KEY_PREFIX)) {
break;

if (recoveryEnabled) {
secretsFile = initRecoveryDb(SECRETS_RECOVERY_FILE_NAME);

// Make sure this is protected in case its not in the NM recovery dir
FileSystem fs = FileSystem.getLocal(_conf);
fs.mkdirs(new Path(secretsFile.getPath()), new FsPermission((short) 0700));

db = LevelDBProvider.initLevelDB(secretsFile, CURRENT_VERSION, mapper);
logger.info("Recovery location is: " + secretsFile.getPath());
if (db != null) {
logger.info("Going to reload spark shuffle data");
DBIterator itr = db.iterator();
itr.seek(APP_CREDS_KEY_PREFIX.getBytes(StandardCharsets.UTF_8));
while (itr.hasNext()) {
Map.Entry<byte[], byte[]> e = itr.next();
String key = new String(e.getKey(), StandardCharsets.UTF_8);
if (!key.startsWith(APP_CREDS_KEY_PREFIX)) {
break;
}
String id = parseDbAppKey(key);
ByteBuffer secret = mapper.readValue(e.getValue(), ByteBuffer.class);
logger.info("Reloading tokens for app: " + id);
secretManager.registerApp(id, secret);
}
String id = parseDbAppKey(key);
ByteBuffer secret = mapper.readValue(e.getValue(), ByteBuffer.class);
logger.info("Reloading tokens for app: " + id);
secretManager.registerApp(id, secret);
}
}
}
Expand Down Expand Up @@ -338,49 +349,41 @@ protected Path getRecoveryPath(String fileName) {
* it will uses a YARN local dir.

@tgravescs tgravescs Aug 28, 2017

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.

need to update comment, probably just remove the last sentence.

*/
protected File initRecoveryDb(String dbName) {
if (_recoveryPath != null) {
File recoveryFile = new File(_recoveryPath.toUri().getPath(), dbName);
if (recoveryFile.exists()) {
return recoveryFile;
}
Preconditions.checkNotNull(_recoveryPath,
"recovery path should not be null if NM recovery is enabled");

File recoveryFile = new File(_recoveryPath.toUri().getPath(), dbName);
if (recoveryFile.exists()) {
return recoveryFile;
}

// db doesn't exist in recovery path go check local dirs for it
String[] localDirs = _conf.getTrimmedStrings("yarn.nodemanager.local-dirs");
for (String dir : localDirs) {
File f = new File(new Path(dir).toUri().getPath(), dbName);
if (f.exists()) {
if (_recoveryPath == null) {
// If NM recovery is not enabled, we should specify the recovery path using NM local
// dirs, which is compatible with the old code.
_recoveryPath = new Path(dir);
return f;
} else {
// If the recovery path is set then either NM recovery is enabled or another recovery
// DB has been initialized. If NM recovery is enabled and had set the recovery path
// make sure to move all DBs to the recovery path from the old NM local dirs.
// If another DB was initialized first just make sure all the DBs are in the same
// location.
Path newLoc = new Path(_recoveryPath, dbName);
Path copyFrom = new Path(f.toURI());
if (!newLoc.equals(copyFrom)) {
logger.info("Moving " + copyFrom + " to: " + newLoc);
try {
// The move here needs to handle moving non-empty directories across NFS mounts
FileSystem fs = FileSystem.getLocal(_conf);
fs.rename(copyFrom, newLoc);
} catch (Exception e) {
// Fail to move recovery file to new path, just continue on with new DB location
logger.error("Failed to move recovery file {} to the path {}",
dbName, _recoveryPath.toString(), e);
}
// If the recovery path is set then either NM recovery is enabled or another recovery
// DB has been initialized. If NM recovery is enabled and had set the recovery path
// make sure to move all DBs to the recovery path from the old NM local dirs.
// If another DB was initialized first just make sure all the DBs are in the same
// location.
Path newLoc = new Path(_recoveryPath, dbName);
Path copyFrom = new Path(f.toURI());
if (!newLoc.equals(copyFrom)) {
logger.info("Moving " + copyFrom + " to: " + newLoc);
try {
// The move here needs to handle moving non-empty directories across NFS mounts
FileSystem fs = FileSystem.getLocal(_conf);
fs.rename(copyFrom, newLoc);
} catch (Exception e) {
// Fail to move recovery file to new path, just continue on with new DB location
logger.error("Failed to move recovery file {} to the path {}",
dbName, _recoveryPath.toString(), e);
}
return new File(newLoc.toUri().getPath());
}
return new File(newLoc.toUri().getPath());
}
}
if (_recoveryPath == null) {
_recoveryPath = new Path(localDirs[0]);
}

return new File(_recoveryPath.toUri().getPath(), dbName);
}
Expand Down
33 changes: 25 additions & 8 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 @@ -70,11 +70,18 @@ class YarnShuffleIntegrationSuite extends BaseYarnClusterSuite {
val finalState = runSpark(
false,
mainClassName(YarnExternalShuffleDriver.getClass),
appArgs = Seq(result.getAbsolutePath(), registeredExecFile.getAbsolutePath),
appArgs = if (registeredExecFile != null) {
Seq(result.getAbsolutePath, registeredExecFile.getAbsolutePath)
} else {
Seq(result.getAbsolutePath)
},
extraConf = extraSparkConf()
)
checkResult(finalState, result)
assert(YarnTestAccessor.getRegisteredExecutorFile(shuffleService).exists())

if (registeredExecFile != null) {
assert(YarnTestAccessor.getRegisteredExecutorFile(shuffleService).exists())
}
}
}

Expand Down Expand Up @@ -105,7 +112,7 @@ private object YarnExternalShuffleDriver extends Logging with Matchers {
val WAIT_TIMEOUT_MILLIS = 10000

def main(args: Array[String]): Unit = {
if (args.length != 2) {
if (args.length > 2) {
// scalastyle:off println
System.err.println(
s"""
Expand All @@ -121,22 +128,32 @@ private object YarnExternalShuffleDriver extends Logging with Matchers {
.setAppName("External Shuffle Test"))
val conf = sc.getConf
val status = new File(args(0))
val registeredExecFile = new File(args(1))
val registeredExecFile = if (args.length == 2) {
new File(args(1))
} else {
null
}
logInfo("shuffle service executor file = " + registeredExecFile)
var result = "failure"
val execStateCopy = new File(registeredExecFile.getAbsolutePath + "_dup")
val execStateCopy = Option(registeredExecFile).map { file =>
new File(file.getAbsolutePath + "_dup")
}.orNull
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
FileUtils.copyDirectory(registeredExecFile, execStateCopy)
assert(!ShuffleTestAccessor.reloadRegisteredExecutors(execStateCopy).isEmpty)
if (registeredExecFile != null && execStateCopy != null) {
FileUtils.copyDirectory(registeredExecFile, execStateCopy)
assert(!ShuffleTestAccessor.reloadRegisteredExecutors(execStateCopy).isEmpty)
}
} finally {
sc.stop()
FileUtils.deleteDirectory(execStateCopy)
if (execStateCopy != null) {
FileUtils.deleteDirectory(execStateCopy)
}
Files.write(result, status, StandardCharsets.UTF_8)
}
}
Expand Down
38 changes: 28 additions & 10 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 @@ -44,6 +44,8 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
private[yarn] var yarnConfig: YarnConfiguration = null
private[yarn] val SORT_MANAGER = "org.apache.spark.shuffle.sort.SortShuffleManager"

private var recoveryLocalDir: File = _

override def beforeEach(): Unit = {
super.beforeEach()
yarnConfig = new YarnConfiguration()
Expand All @@ -54,6 +56,8 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
yarnConfig.setBoolean(YarnShuffleService.STOP_ON_FAILURE_KEY, true)
val localDir = Utils.createTempDir()
yarnConfig.set(YarnConfiguration.NM_LOCAL_DIRS, localDir.getAbsolutePath)

recoveryLocalDir = Utils.createTempDir()
}

var s1: YarnShuffleService = null
Expand All @@ -80,7 +84,9 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
}

test("executor state kept across NM restart") {
yarnConfig.set(YarnConfiguration.NM_RECOVERY_ENABLED, "true")
s1 = new YarnShuffleService
s1.setRecoveryPath(new Path(recoveryLocalDir.toURI))
// set auth to true to test the secrets recovery
yarnConfig.setBoolean(SecurityManager.SPARK_AUTH_CONF, true)
s1.init(yarnConfig)
Expand Down Expand Up @@ -123,6 +129,7 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
// now we pretend the shuffle service goes down, and comes back up
s1.stop()
s2 = new YarnShuffleService
s2.setRecoveryPath(new Path(recoveryLocalDir.toURI))
s2.init(yarnConfig)
s2.secretsFile should be (secretsFile)
s2.registeredExecutorFile should be (execStateFile)
Expand All @@ -140,6 +147,7 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
// Act like the NM restarts one more time
s2.stop()
s3 = new YarnShuffleService
s3.setRecoveryPath(new Path(recoveryLocalDir.toURI))
s3.init(yarnConfig)
s3.registeredExecutorFile should be (execStateFile)
s3.secretsFile should be (secretsFile)
Expand All @@ -155,7 +163,9 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
}

test("removed applications should not be in registered executor file") {
yarnConfig.set(YarnConfiguration.NM_RECOVERY_ENABLED, "true")
s1 = new YarnShuffleService
s1.setRecoveryPath(new Path(recoveryLocalDir.toURI))
yarnConfig.setBoolean(SecurityManager.SPARK_AUTH_CONF, false)
s1.init(yarnConfig)
val secretsFile = s1.secretsFile
Expand Down Expand Up @@ -189,7 +199,9 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
}

test("shuffle service should be robust to corrupt registered executor file") {
yarnConfig.set(YarnConfiguration.NM_RECOVERY_ENABLED, "true")
s1 = new YarnShuffleService
s1.setRecoveryPath(new Path(recoveryLocalDir.toURI))
s1.init(yarnConfig)
val app1Id = ApplicationId.newInstance(0, 1)
val app1Data = makeAppInfo("user", app1Id)
Expand All @@ -215,6 +227,7 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
out.close()

s2 = new YarnShuffleService
s2.setRecoveryPath(new Path(recoveryLocalDir.toURI))
s2.init(yarnConfig)
s2.registeredExecutorFile should be (execStateFile)

Expand All @@ -234,6 +247,7 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd

// another stop & restart should be fine though (eg., we recover from previous corruption)
s3 = new YarnShuffleService
s3.setRecoveryPath(new Path(recoveryLocalDir.toURI))
s3.init(yarnConfig)
s3.registeredExecutorFile should be (execStateFile)
val handler3 = s3.blockHandler
Expand All @@ -247,21 +261,14 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
test("get correct recovery path") {
// Test recovery path is set outside the shuffle service, this is to simulate NM recovery
// enabled scenario, where recovery path will be set by yarn.
yarnConfig.set(YarnConfiguration.NM_RECOVERY_ENABLED, "true")
s1 = new YarnShuffleService
val recoveryPath = new Path(Utils.createTempDir().toURI)
s1.setRecoveryPath(recoveryPath)

s1.init(yarnConfig)
s1._recoveryPath should be (recoveryPath)
s1.stop()

// Test recovery path is set inside the shuffle service, this will be happened when NM
// recovery is not enabled or there's no NM recovery (Hadoop 2.5-).
s2 = new YarnShuffleService
s2.init(yarnConfig)
s2._recoveryPath should be
(new Path(yarnConfig.getTrimmedStrings("yarn.nodemanager.local-dirs")(0)))
s2.stop()
}

test("moving recovery file from NM local dir to recovery path") {
Expand All @@ -270,7 +277,9 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd

// Simulate s1 is running on old version of Hadoop in which recovery file is in the NM local
// dir.
yarnConfig.set(YarnConfiguration.NM_RECOVERY_ENABLED, "true")
s1 = new YarnShuffleService
s1.setRecoveryPath(new Path(yarnConfig.getTrimmedStrings(YarnConfiguration.NM_LOCAL_DIRS)(0)))
// set auth to true to test the secrets recovery
yarnConfig.setBoolean(SecurityManager.SPARK_AUTH_CONF, true)
s1.init(yarnConfig)
Expand Down Expand Up @@ -308,7 +317,7 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd

// Simulate s2 is running on Hadoop 2.5+ with NM recovery is enabled.
assert(execStateFile.exists())
val recoveryPath = new Path(Utils.createTempDir().toURI)
val recoveryPath = new Path(recoveryLocalDir.toURI)
s2 = new YarnShuffleService
s2.setRecoveryPath(recoveryPath)
s2.init(yarnConfig)
Expand Down Expand Up @@ -347,10 +356,11 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
// Set up a read-only local dir.
val roDir = Utils.createTempDir()
Files.setPosixFilePermissions(roDir.toPath(), EnumSet.of(OWNER_READ, OWNER_EXECUTE))
yarnConfig.set(YarnConfiguration.NM_LOCAL_DIRS, roDir.getAbsolutePath())
yarnConfig.set(YarnConfiguration.NM_RECOVERY_ENABLED, "true")

// Try to start the shuffle service, it should fail.
val service = new YarnShuffleService()
service.setRecoveryPath(new Path(roDir.toURI))

try {
val error = intercept[ServiceStateException] {
Expand All @@ -369,4 +379,12 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
new ApplicationInitializationContext(user, appId, secret)
}

test("recovery db should not be created if NM recovery is not enabled") {
s1 = new YarnShuffleService
s1.init(yarnConfig)
s1._recoveryPath should be (null)
s1.registeredExecutorFile should be (null)
s1.secretsFile should be (null)
}

}