Author: sharad
Date: Wed Mar 30 09:48:13 2011
New Revision: 1086882
URL: http://svn.apache.org/viewvc?rev=1086882&view=rev
Log:
Fixed paths in History server and integrated with MiniMRYarnCluster.
Contributed by Krishna Ramachandran.
Added:
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/TestMRJobsWithHistoryService.java
Modified:
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/MiniMRYarnCluster.java
Modified:
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java?rev=1086882&r1=1086881&r2=1086882&view=diff
==============================================================================
---
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java
(original)
+++
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java
Wed Mar 30 09:48:13 2011
@@ -35,8 +35,10 @@ import org.apache.hadoop.mapreduce.TypeC
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser;
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.JobInfo;
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.TaskInfo;
+import org.apache.hadoop.mapreduce.v2.YarnMRJobConfig;
import org.apache.hadoop.mapreduce.v2.app.job.Task;
import org.apache.hadoop.yarn.YarnException;
+import org.apache.hadoop.yarn.conf.YARNApplicationConstants;
import org.apache.hadoop.mapreduce.v2.api.Counters;
import org.apache.hadoop.mapreduce.v2.api.JobID;
import org.apache.hadoop.mapreduce.v2.api.JobReport;
@@ -158,22 +160,26 @@ public class CompletedJob implements org
LOG.error("user null is not allowed");
}
String jobName = TypeConverter.fromYarn(jobID).toString();
- String jobhistoryDir = conf.get("yarn.server.nodemanager.jobhistory",
- "file:///tmp/yarn/done")
+ String defaultDoneDir = conf.get(
+ YARNApplicationConstants.APPS_STAGING_DIR_KEY) + "/history/done";
+ String jobhistoryDir =
+ conf.get(YarnMRJobConfig.HISTORY_DONE_DIR_KEY, defaultDoneDir)
+ "/" + user;
FSDataInputStream in = null;
- String jobhistoryFileName = jobName; // TODO use existing hadoop dire
- // structure
- Path historyFilePath = new Path(jobhistoryDir, jobhistoryFileName);
-
+ Path historyFile = null;
try {
- FileContext fc = FileContext.getFileContext(historyFilePath.toUri());
- in = fc.open(historyFilePath);
+ Path doneDirPath = FileContext.getFileContext(conf).makeQualified(
+ new Path(jobhistoryDir));
+ FileContext fc =
+ FileContext.getFileContext(doneDirPath.toUri(),conf);
+ historyFile =
+ fc.makeQualified(new Path(doneDirPath, jobName));
+ in = fc.open(historyFile);
JobHistoryParser parser = new JobHistoryParser(in);
jobInfo = parser.parse();
LOG.info("jobInfo loaded");
} catch (IOException e) {
- throw new YarnException("Could not load history file " + historyFilePath,
+ throw new YarnException("Could not load history file " + historyFile,
e);
}
@@ -193,8 +199,6 @@ public class CompletedJob implements org
// TODO: populate the TaskAttemptCompletionEvent
completionEvents = new TaskAttemptCompletionEvent[0];
-
-
}
@Override
Modified:
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/MiniMRYarnCluster.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/MiniMRYarnCluster.java?rev=1086882&r1=1086881&r2=1086882&view=diff
==============================================================================
---
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/MiniMRYarnCluster.java
(original)
+++
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/MiniMRYarnCluster.java
Wed Mar 30 09:48:13 2011
@@ -50,12 +50,13 @@ public class MiniMRYarnCluster extends M
private static final Log LOG = LogFactory.getLog(MiniMRYarnCluster.class);
private JobHistoryServer historyServer;
+ private JobHistoryServerWrapper historyServerWrapper;
public MiniMRYarnCluster(String testName) {
super(testName);
//TODO: add the history server
- //historyServer = new JobHistoryServerWrapper();
- //addService(historyServer);
+ historyServerWrapper = new JobHistoryServerWrapper();
+ addService(historyServerWrapper);
}
@Override
@@ -70,10 +71,6 @@ public class MiniMRYarnCluster extends M
"apps_staging_dir/${user.name}/").getAbsolutePath());
conf.set(MRConfig.MASTER_ADDRESS, "test"); // The default is local because
of
// which shuffle doesn't happen
- conf.set(YarnMRJobConfig.HISTORY_STAGING_DIR_KEY,
- "file:///tmp/yarn/");
- conf.set(YarnMRJobConfig.HISTORY_DONE_DIR_KEY,
- "file:///tmp/yarn/done/");
//configure the shuffle service in NM
conf.setStrings(AuxServices.AUX_SERVICES,
new String[] { ShuffleHandler.MAPREDUCE_SHUFFLE_SERVICEID });
@@ -111,4 +108,7 @@ public class MiniMRYarnCluster extends M
}
}
}
+ public JobHistoryServer getHistoryServer() {
+ return this.historyServer;
+ }
}
Added:
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/TestMRJobsWithHistoryService.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/TestMRJobsWithHistoryService.java?rev=1086882&view=auto
==============================================================================
---
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/TestMRJobsWithHistoryService.java
(added)
+++
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/v2/TestMRJobsWithHistoryService.java
Wed Mar 30 09:48:13 2011
@@ -0,0 +1,116 @@
+/**
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.hadoop.mapreduce.v2;
+
+import java.io.File;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import org.apache.avro.ipc.AvroRemoteException;
+
+import junit.framework.Assert;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.FailingMapper;
+import org.apache.hadoop.RandomTextWriterJob;
+import org.apache.hadoop.SleepJob;
+import org.apache.hadoop.RandomTextWriterJob.RandomInputFormat;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.YARNRunner;
+import org.apache.hadoop.mapreduce.Counters;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobStatus;
+import org.apache.hadoop.mapreduce.MRJobConfig;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.TaskCompletionEvent;
+import org.apache.hadoop.mapreduce.TaskID;
+import org.apache.hadoop.mapreduce.TaskType;
+import org.apache.hadoop.mapreduce.TypeConverter;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.apache.hadoop.yarn.ApplicationID;
+import org.apache.hadoop.yarn.ApplicationState;
+import org.apache.hadoop.yarn.conf.YARNApplicationConstants;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.YarnServerConfig;
+import org.apache.hadoop.yarn.server.nodemanager.NMConfig;
+import org.apache.hadoop.yarn.server.resourcemanager.RMConfig;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class TestMRJobsWithHistoryService {
+
+ private static final Log LOG = LogFactory.getLog(TestMRJobs.class);
+
+ private static MiniMRYarnCluster mrCluster;
+
+ @Before
+ public void setup() throws InterruptedException, IOException {
+
+ if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
+ LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR + " not found. Not
running test.");
+ return;
+ }
+
+ if (mrCluster == null) {
+ mrCluster = new MiniMRYarnCluster(getClass().getName());
+ mrCluster.init(new Configuration());
+ mrCluster.start();
+ }
+ }
+
+ @Test
+ public void testJobHistoryData() throws IOException, InterruptedException,
+ AvroRemoteException, ClassNotFoundException {
+ if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
+ LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ + " not found. Not running test.");
+ return;
+ }
+
+ SleepJob sleepJob = new SleepJob();
+ sleepJob.setConf(mrCluster.getConfig());
+ // Job with 3 maps and 2 reduces
+ Job job = sleepJob.createJob(3, 2, 1000, 1, 500, 1);
+ job.setJar(new File(MiniMRYarnCluster.APPJAR).getAbsolutePath());
+ job.waitForCompletion(true);
+ Counters counterMR = job.getCounters();
+ ApplicationID appID = TypeConverter.toYarn(job.getJobID()).appID;
+ while (true) {
+ Thread.sleep(1000);
+ if (mrCluster.getResourceManager().getApplicationsManager()
+ .getApplication(appID).state().equals(ApplicationState.COMPLETED))
+ break;
+ }
+ Counters counterHS = job.getCounters();
+ //TODO the Assert below worked. need to check
+ //Should we compare each field or convert to V2 counter and compare
+ LOG.info("CounterHS " + counterHS);
+ LOG.info("CounterMR " + counterMR);
+ Assert.assertEquals(counterHS, counterMR);
+ }
+
+}
