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 sj...@apache.org on 2016/07/20 03:18:44 UTC
hadoop git commit: MAPREDUCE-6365. Refactor
JobResourceUploader#uploadFilesInternal (Chris Trezzo via sjlee)
Repository: hadoop
Updated Branches:
refs/heads/trunk dc065dd64 -> 8f0d3d69d
MAPREDUCE-6365. Refactor JobResourceUploader#uploadFilesInternal (Chris Trezzo via sjlee)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/8f0d3d69
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/8f0d3d69
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/8f0d3d69
Branch: refs/heads/trunk
Commit: 8f0d3d69d65a252439610e6f13d679808d768569
Parents: dc065dd
Author: Sangjin Lee <sj...@apache.org>
Authored: Tue Jul 19 20:15:37 2016 -0700
Committer: Sangjin Lee <sj...@apache.org>
Committed: Tue Jul 19 20:15:37 2016 -0700
----------------------------------------------------------------------
.../hadoop/mapreduce/JobResourceUploader.java | 62 ++++++++++++--------
.../apache/hadoop/mapreduce/JobSubmitter.java | 2 +-
2 files changed, 40 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f0d3d69/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobResourceUploader.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobResourceUploader.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobResourceUploader.java
index 90e1fba..fa4dd86 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobResourceUploader.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobResourceUploader.java
@@ -54,7 +54,7 @@ class JobResourceUploader {
* @param submitJobDir the submission directory of the job
* @throws IOException
*/
- public void uploadFiles(Job job, Path submitJobDir) throws IOException {
+ public void uploadResources(Job job, Path submitJobDir) throws IOException {
Configuration conf = job.getConfiguration();
short replication =
(short) conf.getInt(Job.SUBMIT_REPLICATION,
@@ -66,12 +66,6 @@ class JobResourceUploader {
+ "with ToolRunner to remedy this.");
}
- // get all the command line arguments passed in by the user conf
- String files = conf.get("tmpfiles");
- String libjars = conf.get("tmpjars");
- String archives = conf.get("tmparchives");
- String jobJar = job.getJar();
-
//
// Figure out what fs the JobTracker is using. Copy the
// job to it, under a temporary name. This allows DFS to work,
@@ -92,12 +86,27 @@ class JobResourceUploader {
FsPermission mapredSysPerms =
new FsPermission(JobSubmissionFiles.JOB_DIR_PERMISSION);
FileSystem.mkdirs(jtFs, submitJobDir, mapredSysPerms);
- Path filesDir = JobSubmissionFiles.getJobDistCacheFiles(submitJobDir);
- Path archivesDir = JobSubmissionFiles.getJobDistCacheArchives(submitJobDir);
- Path libjarsDir = JobSubmissionFiles.getJobDistCacheLibjars(submitJobDir);
// add all the command line files/ jars and archive
// first copy them to jobtrackers filesystem
+ uploadFiles(conf, submitJobDir, mapredSysPerms, replication);
+ uploadLibJars(conf, submitJobDir, mapredSysPerms, replication);
+ uploadArchives(conf, submitJobDir, mapredSysPerms, replication);
+ uploadJobJar(job, submitJobDir, replication);
+ addLog4jToDistributedCache(job, submitJobDir);
+
+ // set the timestamps of the archives and files
+ // set the public/private visibility of the archives and files
+ ClientDistributedCacheManager.determineTimestampsAndCacheVisibilities(conf);
+ // get DelegationToken for cached file
+ ClientDistributedCacheManager.getDelegationTokens(conf,
+ job.getCredentials());
+ }
+
+ private void uploadFiles(Configuration conf, Path submitJobDir,
+ FsPermission mapredSysPerms, short submitReplication) throws IOException {
+ String files = conf.get("tmpfiles");
+ Path filesDir = JobSubmissionFiles.getJobDistCacheFiles(submitJobDir);
if (files != null) {
FileSystem.mkdirs(jtFs, filesDir, mapredSysPerms);
String[] fileArr = files.split(",");
@@ -109,7 +118,7 @@ class JobResourceUploader {
throw new IllegalArgumentException(e);
}
Path tmp = new Path(tmpURI);
- Path newPath = copyRemoteFiles(filesDir, tmp, conf, replication);
+ Path newPath = copyRemoteFiles(filesDir, tmp, conf, submitReplication);
try {
URI pathURI = getPathURI(newPath, tmpURI.getFragment());
DistributedCache.addCacheFile(pathURI, conf);
@@ -119,13 +128,19 @@ class JobResourceUploader {
}
}
}
+ }
+ private void uploadLibJars(Configuration conf, Path submitJobDir,
+ FsPermission mapredSysPerms, short submitReplication) throws IOException {
+ String libjars = conf.get("tmpjars");
+ Path libjarsDir = JobSubmissionFiles.getJobDistCacheLibjars(submitJobDir);
if (libjars != null) {
FileSystem.mkdirs(jtFs, libjarsDir, mapredSysPerms);
String[] libjarsArr = libjars.split(",");
for (String tmpjars : libjarsArr) {
Path tmp = new Path(tmpjars);
- Path newPath = copyRemoteFiles(libjarsDir, tmp, conf, replication);
+ Path newPath =
+ copyRemoteFiles(libjarsDir, tmp, conf, submitReplication);
// Add each file to the classpath
DistributedCache.addFileToClassPath(
@@ -140,7 +155,12 @@ class JobResourceUploader {
DistributedCache.addCacheFile(libJarsDirWildcard.toUri(), conf);
}
}
+ }
+ private void uploadArchives(Configuration conf, Path submitJobDir,
+ FsPermission mapredSysPerms, short submitReplication) throws IOException {
+ String archives = conf.get("tmparchives");
+ Path archivesDir = JobSubmissionFiles.getJobDistCacheArchives(submitJobDir);
if (archives != null) {
FileSystem.mkdirs(jtFs, archivesDir, mapredSysPerms);
String[] archivesArr = archives.split(",");
@@ -152,7 +172,8 @@ class JobResourceUploader {
throw new IllegalArgumentException(e);
}
Path tmp = new Path(tmpURI);
- Path newPath = copyRemoteFiles(archivesDir, tmp, conf, replication);
+ Path newPath =
+ copyRemoteFiles(archivesDir, tmp, conf, submitReplication);
try {
URI pathURI = getPathURI(newPath, tmpURI.getFragment());
DistributedCache.addCacheArchive(pathURI, conf);
@@ -162,7 +183,11 @@ class JobResourceUploader {
}
}
}
+ }
+ private void uploadJobJar(Job job, Path submitJobDir, short submitReplication)
+ throws IOException {
+ String jobJar = job.getJar();
if (jobJar != null) { // copy jar to JobTracker's fs
// use jar name if job is not named.
if ("".equals(job.getJobName())) {
@@ -174,22 +199,13 @@ class JobResourceUploader {
// we don't need to copy it from local fs
if (jobJarURI.getScheme() == null || jobJarURI.getScheme().equals("file")) {
copyJar(jobJarPath, JobSubmissionFiles.getJobJar(submitJobDir),
- replication);
+ submitReplication);
job.setJar(JobSubmissionFiles.getJobJar(submitJobDir).toString());
}
} else {
LOG.warn("No job jar file set. User classes may not be found. "
+ "See Job or Job#setJar(String).");
}
-
- addLog4jToDistributedCache(job, submitJobDir);
-
- // set the timestamps of the archives and files
- // set the public/private visibility of the archives and files
- ClientDistributedCacheManager.determineTimestampsAndCacheVisibilities(conf);
- // get DelegationToken for cached file
- ClientDistributedCacheManager.getDelegationTokens(conf,
- job.getCredentials());
}
// copies a file to the jobtracker filesystem and returns the path where it
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f0d3d69/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobSubmitter.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobSubmitter.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobSubmitter.java
index 22874e1..d1cc72c 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobSubmitter.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobSubmitter.java
@@ -99,7 +99,7 @@ class JobSubmitter {
Job.DEFAULT_USE_WILDCARD_FOR_LIBJARS);
JobResourceUploader rUploader = new JobResourceUploader(jtFs, useWildcards);
- rUploader.uploadFiles(job, jobSubmitDir);
+ rUploader.uploadResources(job, jobSubmitDir);
// Get the working directory. If not set, sets it to filesystem working dir
// This code has been added so that working directory reset before running
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org