You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by su...@apache.org on 2017/04/03 20:58:06 UTC
[12/50] [abbrv] hadoop git commit: MAPREDUCE-6824.
TaskAttemptImpl#createCommonContainerLaunchContext is longer than 150 lines.
Contributed by Chris Trezzo.
MAPREDUCE-6824. TaskAttemptImpl#createCommonContainerLaunchContext is longer than 150 lines. Contributed by Chris Trezzo.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/845529b3
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/845529b3
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/845529b3
Branch: refs/heads/YARN-2915
Commit: 845529b3ab338e759665a687eb525fb2cccde7bf
Parents: a4b5aa8
Author: Akira Ajisaka <aa...@apache.org>
Authored: Mon Apr 3 13:06:24 2017 +0900
Committer: Akira Ajisaka <aa...@apache.org>
Committed: Mon Apr 3 13:06:54 2017 +0900
----------------------------------------------------------------------
.../v2/app/job/impl/TaskAttemptImpl.java | 285 ++++++++++---------
1 file changed, 153 insertions(+), 132 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/845529b3/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
index 4305824..9ea1b9a 100755
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
@@ -755,7 +755,7 @@ public abstract class TaskAttemptImpl implements
new HashMap<String, LocalResource>();
// Application environment
- Map<String, String> environment = new HashMap<String, String>();
+ Map<String, String> environment;
// Service data
Map<String, ByteBuffer> serviceData = new HashMap<String, ByteBuffer>();
@@ -763,157 +763,178 @@ public abstract class TaskAttemptImpl implements
// Tokens
ByteBuffer taskCredentialsBuffer = ByteBuffer.wrap(new byte[]{});
try {
- FileSystem remoteFS = FileSystem.get(conf);
-
- // //////////// Set up JobJar to be localized properly on the remote NM.
- String jobJar = conf.get(MRJobConfig.JAR);
- if (jobJar != null) {
- final Path jobJarPath = new Path(jobJar);
- final FileSystem jobJarFs = FileSystem.get(jobJarPath.toUri(), conf);
- Path remoteJobJar = jobJarPath.makeQualified(jobJarFs.getUri(),
- jobJarFs.getWorkingDirectory());
- LocalResource rc = createLocalResource(jobJarFs, remoteJobJar,
- LocalResourceType.PATTERN, LocalResourceVisibility.APPLICATION);
- String pattern = conf.getPattern(JobContext.JAR_UNPACK_PATTERN,
- JobConf.UNPACK_JAR_PATTERN_DEFAULT).pattern();
- rc.setPattern(pattern);
- localResources.put(MRJobConfig.JOB_JAR, rc);
- LOG.info("The job-jar file on the remote FS is "
- + remoteJobJar.toUri().toASCIIString());
- } else {
- // Job jar may be null. For e.g, for pipes, the job jar is the hadoop
- // mapreduce jar itself which is already on the classpath.
- LOG.info("Job jar is not present. "
- + "Not adding any jar to the list of resources.");
- }
- // //////////// End of JobJar setup
-
- // //////////// Set up JobConf to be localized properly on the remote NM.
- Path path =
- MRApps.getStagingAreaDir(conf, UserGroupInformation
- .getCurrentUser().getShortUserName());
- Path remoteJobSubmitDir =
- new Path(path, oldJobId.toString());
- Path remoteJobConfPath =
- new Path(remoteJobSubmitDir, MRJobConfig.JOB_CONF_FILE);
- localResources.put(
- MRJobConfig.JOB_CONF_FILE,
- createLocalResource(remoteFS, remoteJobConfPath,
- LocalResourceType.FILE, LocalResourceVisibility.APPLICATION));
- LOG.info("The job-conf file on the remote FS is "
- + remoteJobConfPath.toUri().toASCIIString());
- // //////////// End of JobConf setup
- // Setup DistributedCache
- MRApps.setupDistributedCache(conf, localResources);
+ configureJobJar(conf, localResources);
- // Setup up task credentials buffer
- LOG.info("Adding #" + credentials.numberOfTokens()
- + " tokens and #" + credentials.numberOfSecretKeys()
- + " secret keys for NM use for launching container");
- Credentials taskCredentials = new Credentials(credentials);
+ configureJobConf(conf, localResources, oldJobId);
- // LocalStorageToken is needed irrespective of whether security is enabled
- // or not.
- TokenCache.setJobToken(jobToken, taskCredentials);
+ // Setup DistributedCache
+ MRApps.setupDistributedCache(conf, localResources);
- DataOutputBuffer containerTokens_dob = new DataOutputBuffer();
- LOG.info("Size of containertokens_dob is "
- + taskCredentials.numberOfTokens());
- taskCredentials.writeTokenStorageToStream(containerTokens_dob);
taskCredentialsBuffer =
- ByteBuffer.wrap(containerTokens_dob.getData(), 0,
- containerTokens_dob.getLength());
-
- // Add shuffle secret key
- // The secret key is converted to a JobToken to preserve backwards
- // compatibility with an older ShuffleHandler running on an NM.
- LOG.info("Putting shuffle token in serviceData");
- byte[] shuffleSecret = TokenCache.getShuffleSecretKey(credentials);
- if (shuffleSecret == null) {
- LOG.warn("Cannot locate shuffle secret in credentials."
- + " Using job token as shuffle secret.");
- shuffleSecret = jobToken.getPassword();
- }
- Token<JobTokenIdentifier> shuffleToken = new Token<JobTokenIdentifier>(
- jobToken.getIdentifier(), shuffleSecret, jobToken.getKind(),
- jobToken.getService());
- serviceData.put(ShuffleHandler.MAPREDUCE_SHUFFLE_SERVICEID,
- ShuffleHandler.serializeServiceData(shuffleToken));
-
- // add external shuffle-providers - if any
- Collection<String> shuffleProviders = conf.getStringCollection(
- MRJobConfig.MAPREDUCE_JOB_SHUFFLE_PROVIDER_SERVICES);
- if (! shuffleProviders.isEmpty()) {
- Collection<String> auxNames = conf.getStringCollection(
- YarnConfiguration.NM_AUX_SERVICES);
-
- for (final String shuffleProvider : shuffleProviders) {
- if (shuffleProvider.equals(ShuffleHandler.MAPREDUCE_SHUFFLE_SERVICEID)) {
- continue; // skip built-in shuffle-provider that was already inserted with shuffle secret key
- }
- if (auxNames.contains(shuffleProvider)) {
- LOG.info("Adding ShuffleProvider Service: " + shuffleProvider + " to serviceData");
- // This only serves for INIT_APP notifications
- // The shuffle service needs to be able to work with the host:port information provided by the AM
- // (i.e. shuffle services which require custom location / other configuration are not supported)
- serviceData.put(shuffleProvider, ByteBuffer.allocate(0));
- }
- else {
- throw new YarnRuntimeException("ShuffleProvider Service: " + shuffleProvider +
- " was NOT found in the list of aux-services that are available in this NM." +
- " You may need to specify this ShuffleProvider as an aux-service in your yarn-site.xml");
- }
- }
- }
+ configureTokens(jobToken, credentials, serviceData);
- MRApps.addToEnvironment(
- environment,
- Environment.CLASSPATH.name(),
- getInitialClasspath(conf), conf);
+ addExternalShuffleProviders(conf, serviceData);
+
+ environment = configureEnv(conf);
- if (initialAppClasspath != null) {
- MRApps.addToEnvironment(
- environment,
- Environment.APP_CLASSPATH.name(),
- initialAppClasspath, conf);
- }
} catch (IOException e) {
throw new YarnRuntimeException(e);
}
- // Shell
- environment.put(
- Environment.SHELL.name(),
- conf.get(
- MRJobConfig.MAPRED_ADMIN_USER_SHELL,
- MRJobConfig.DEFAULT_SHELL)
- );
-
- // Add pwd to LD_LIBRARY_PATH, add this before adding anything else
- MRApps.addToEnvironment(
- environment,
- Environment.LD_LIBRARY_PATH.name(),
- MRApps.crossPlatformifyMREnv(conf, Environment.PWD), conf);
-
- // Add the env variables passed by the admin
- MRApps.setEnvFromInputString(
- environment,
- conf.get(
- MRJobConfig.MAPRED_ADMIN_USER_ENV,
- MRJobConfig.DEFAULT_MAPRED_ADMIN_USER_ENV), conf
- );
-
// Construct the actual Container
// The null fields are per-container and will be constructed for each
// container separately.
ContainerLaunchContext container =
ContainerLaunchContext.newInstance(localResources, environment, null,
- serviceData, taskCredentialsBuffer, applicationACLs);
+ serviceData, taskCredentialsBuffer, applicationACLs);
return container;
}
+ private static Map<String, String> configureEnv(Configuration conf)
+ throws IOException {
+ Map<String, String> environment = new HashMap<String, String>();
+ MRApps.addToEnvironment(environment, Environment.CLASSPATH.name(),
+ getInitialClasspath(conf), conf);
+
+ if (initialAppClasspath != null) {
+ MRApps.addToEnvironment(environment, Environment.APP_CLASSPATH.name(),
+ initialAppClasspath, conf);
+ }
+
+ // Shell
+ environment.put(Environment.SHELL.name(), conf
+ .get(MRJobConfig.MAPRED_ADMIN_USER_SHELL, MRJobConfig.DEFAULT_SHELL));
+
+ // Add pwd to LD_LIBRARY_PATH, add this before adding anything else
+ MRApps.addToEnvironment(environment, Environment.LD_LIBRARY_PATH.name(),
+ MRApps.crossPlatformifyMREnv(conf, Environment.PWD), conf);
+
+ // Add the env variables passed by the admin
+ MRApps.setEnvFromInputString(environment,
+ conf.get(MRJobConfig.MAPRED_ADMIN_USER_ENV,
+ MRJobConfig.DEFAULT_MAPRED_ADMIN_USER_ENV),
+ conf);
+ return environment;
+ }
+
+ private static void configureJobJar(Configuration conf,
+ Map<String, LocalResource> localResources) throws IOException {
+ // Set up JobJar to be localized properly on the remote NM.
+ String jobJar = conf.get(MRJobConfig.JAR);
+ if (jobJar != null) {
+ final Path jobJarPath = new Path(jobJar);
+ final FileSystem jobJarFs = FileSystem.get(jobJarPath.toUri(), conf);
+ Path remoteJobJar = jobJarPath.makeQualified(jobJarFs.getUri(),
+ jobJarFs.getWorkingDirectory());
+ LocalResource rc = createLocalResource(jobJarFs, remoteJobJar,
+ LocalResourceType.PATTERN, LocalResourceVisibility.APPLICATION);
+ String pattern = conf.getPattern(JobContext.JAR_UNPACK_PATTERN,
+ JobConf.UNPACK_JAR_PATTERN_DEFAULT).pattern();
+ rc.setPattern(pattern);
+ localResources.put(MRJobConfig.JOB_JAR, rc);
+ LOG.info("The job-jar file on the remote FS is "
+ + remoteJobJar.toUri().toASCIIString());
+ } else {
+ // Job jar may be null. For e.g, for pipes, the job jar is the hadoop
+ // mapreduce jar itself which is already on the classpath.
+ LOG.info("Job jar is not present. "
+ + "Not adding any jar to the list of resources.");
+ }
+ }
+
+ private static void configureJobConf(Configuration conf,
+ Map<String, LocalResource> localResources,
+ final org.apache.hadoop.mapred.JobID oldJobId) throws IOException {
+ // Set up JobConf to be localized properly on the remote NM.
+ Path path = MRApps.getStagingAreaDir(conf,
+ UserGroupInformation.getCurrentUser().getShortUserName());
+ Path remoteJobSubmitDir = new Path(path, oldJobId.toString());
+ Path remoteJobConfPath =
+ new Path(remoteJobSubmitDir, MRJobConfig.JOB_CONF_FILE);
+ FileSystem remoteFS = FileSystem.get(conf);
+ localResources.put(MRJobConfig.JOB_CONF_FILE,
+ createLocalResource(remoteFS, remoteJobConfPath, LocalResourceType.FILE,
+ LocalResourceVisibility.APPLICATION));
+ LOG.info("The job-conf file on the remote FS is "
+ + remoteJobConfPath.toUri().toASCIIString());
+ }
+
+ private static ByteBuffer configureTokens(Token<JobTokenIdentifier> jobToken,
+ Credentials credentials,
+ Map<String, ByteBuffer> serviceData) throws IOException {
+ // Setup up task credentials buffer
+ LOG.info("Adding #" + credentials.numberOfTokens() + " tokens and #"
+ + credentials.numberOfSecretKeys()
+ + " secret keys for NM use for launching container");
+ Credentials taskCredentials = new Credentials(credentials);
+
+ // LocalStorageToken is needed irrespective of whether security is enabled
+ // or not.
+ TokenCache.setJobToken(jobToken, taskCredentials);
+
+ DataOutputBuffer containerTokens_dob = new DataOutputBuffer();
+ LOG.info(
+ "Size of containertokens_dob is " + taskCredentials.numberOfTokens());
+ taskCredentials.writeTokenStorageToStream(containerTokens_dob);
+ ByteBuffer taskCredentialsBuffer =
+ ByteBuffer.wrap(containerTokens_dob.getData(), 0,
+ containerTokens_dob.getLength());
+
+ // Add shuffle secret key
+ // The secret key is converted to a JobToken to preserve backwards
+ // compatibility with an older ShuffleHandler running on an NM.
+ LOG.info("Putting shuffle token in serviceData");
+ byte[] shuffleSecret = TokenCache.getShuffleSecretKey(credentials);
+ if (shuffleSecret == null) {
+ LOG.warn("Cannot locate shuffle secret in credentials."
+ + " Using job token as shuffle secret.");
+ shuffleSecret = jobToken.getPassword();
+ }
+ Token<JobTokenIdentifier> shuffleToken =
+ new Token<JobTokenIdentifier>(jobToken.getIdentifier(), shuffleSecret,
+ jobToken.getKind(), jobToken.getService());
+ serviceData.put(ShuffleHandler.MAPREDUCE_SHUFFLE_SERVICEID,
+ ShuffleHandler.serializeServiceData(shuffleToken));
+ return taskCredentialsBuffer;
+ }
+
+ private static void addExternalShuffleProviders(Configuration conf,
+ Map<String, ByteBuffer> serviceData) {
+ // add external shuffle-providers - if any
+ Collection<String> shuffleProviders = conf.getStringCollection(
+ MRJobConfig.MAPREDUCE_JOB_SHUFFLE_PROVIDER_SERVICES);
+ if (!shuffleProviders.isEmpty()) {
+ Collection<String> auxNames =
+ conf.getStringCollection(YarnConfiguration.NM_AUX_SERVICES);
+
+ for (final String shuffleProvider : shuffleProviders) {
+ if (shuffleProvider
+ .equals(ShuffleHandler.MAPREDUCE_SHUFFLE_SERVICEID)) {
+ continue; // skip built-in shuffle-provider that was already inserted
+ // with shuffle secret key
+ }
+ if (auxNames.contains(shuffleProvider)) {
+ LOG.info("Adding ShuffleProvider Service: " + shuffleProvider
+ + " to serviceData");
+ // This only serves for INIT_APP notifications
+ // The shuffle service needs to be able to work with the host:port
+ // information provided by the AM
+ // (i.e. shuffle services which require custom location / other
+ // configuration are not supported)
+ serviceData.put(shuffleProvider, ByteBuffer.allocate(0));
+ } else {
+ throw new YarnRuntimeException("ShuffleProvider Service: "
+ + shuffleProvider
+ + " was NOT found in the list of aux-services that are "
+ + "available in this NM. You may need to specify this "
+ + "ShuffleProvider as an aux-service in your yarn-site.xml");
+ }
+ }
+ }
+ }
+
static ContainerLaunchContext createContainerLaunchContext(
Map<ApplicationAccessType, String> applicationACLs,
Configuration conf, Token<JobTokenIdentifier> jobToken, Task remoteTask,
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org