Author: omalley
Date: Tue Mar 8 06:00:35 2011
New Revision: 1079252
URL: http://svn.apache.org/viewvc?rev=1079252&view=rev
Log:
commit 2df233e2e62b7ec244700f0082c88b11468dabf4
Author: Greg Roelofs <[email protected]>
Date: Tue Jan 25 14:19:57 2011 -0800
"Final" fixes for first round of 42(ish) unit tests, including fixes for
two that were previously incorrectly marked fixed (TestMapReduceLazyOutput,
TestCopyFiles). This does not yet include fixes for newly discovered
TestLazyOutput or TestUserLogCleanup uber-only failures, nor for mixed-
mode TestJobQueueInformation (i.e., uber "failed", non-uber "errored").
It also doesn't include fixes per Dick's review feedback for v10 and v11
patches.
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestMapReduceLazyOutput.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestCopyFiles.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHadoopArchives.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHarFileSystem.java
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java?rev=1079252&r1=1079251&r2=1079252&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java
Tue Mar 8 06:00:35 2011
@@ -673,9 +673,13 @@ public class JobInProgress {
sysConf.getLong("dfs.block.size", 64*1024*1024));
long sysMemSizeForUberSlot = JobTracker.getMemSizeForReduceSlot();
- // user has overall veto power over uberization, or user can set more
+ // User has overall veto power over uberization, or user can set more
// stringent limits than the system specifies, but user may not exceed
- // system limits (for now, anyway)
+ // system limits (for now, anyway). Note that ChainMapper/Reducer are
+ // fundamentally incompatible with UberTask; they employ a blocking
+ // queue between the maps/reduces and thus require parallel execution,
+ // while UberTask loops over maps (and optional reduce) and thus requires
+ // sequential execution.
uberMode = conf.getBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, true)
&& numMapTasks > 0 // temporary restriction until can test reduce-only
&& numMapTasks <= Math.min(sysMaxMaps,
@@ -686,7 +690,13 @@ public class JobInProgress {
conf.getLong(MRJobConfig.JOB_UBERTASK_MAXBYTES, sysMaxBytes))
// ignoring overhead due to UberTask and statics as negligible here:
&& (Math.max(memoryPerMap, memoryPerReduce) <= sysMemSizeForUberSlot
- || sysMemSizeForUberSlot == JobConf.DISABLED_MEMORY_LIMIT);
+ || sysMemSizeForUberSlot == JobConf.DISABLED_MEMORY_LIMIT)
+ && (conf.get(MRJobConfig.MAP_CLASS_ATTR) == null ||
+ conf.get(MRJobConfig.MAP_CLASS_ATTR).lastIndexOf("ChainMapper")
+ == -1)
+ && (conf.get(MRJobConfig.REDUCE_CLASS_ATTR) == null ||
+ conf.get(MRJobConfig.REDUCE_CLASS_ATTR).lastIndexOf("ChainReducer")
+ == -1);
if (isUber()) {
// save internal details for UI and abort-cleanup
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java?rev=1079252&r1=1079251&r2=1079252&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java
Tue Mar 8 06:00:35 2011
@@ -476,7 +476,7 @@ abstract public class Task implements Wr
public abstract boolean isMapTask();
/**
- * Is this really a combo-task masquerading as a plain MapTask?
+ * Is this really a combo-task masquerading as a plain ReduceTask?
*/
public abstract boolean isUberTask();
@@ -498,7 +498,13 @@ abstract public class Task implements Wr
ClassNotFoundException,
InterruptedException {
jobContext = new JobContextImpl(job, id, reporter);
- taskContext = new TaskAttemptContextImpl(job, taskId, reporter);
+ // taskIdForUmbilical is required here since it ultimately determines the
+ // name of the pre-commit HDFS working directory (_temporary/_attempt_xxx),
+ // and anything put in an "_m_" directory will fail to get saved (i.e.,
+ // moved up two levels) in UberTask mode. This mostly (only?) affects
+ // users of HadoopArchives (har) and IndexUpdateOutputFormat (via the
+ // getWorkOutputPath() method in FileOutputFormat).
+ taskContext = new TaskAttemptContextImpl(job, taskIdForUmbilical,
reporter);
if (getState() == TaskStatus.State.UNASSIGNED) {
setState(TaskStatus.State.RUNNING);
}
@@ -512,6 +518,7 @@ abstract public class Task implements Wr
} else {
committer = conf.getOutputCommitter();
}
+ // this will typically be on HDFS, not the node's local filesystem:
Path outputPath = FileOutputFormat.getOutputPath(conf);
if (outputPath != null) {
if ((committer instanceof FileOutputCommitter)) {
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java?rev=1079252&r1=1079251&r2=1079252&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java
Tue Mar 8 06:00:35 2011
@@ -213,6 +213,17 @@ class UberTask extends Task {
return mapIds;
}
+ /**
+ * Within the _local_ filesystem (not HDFS), all activity takes place within
+ * a single directory (e.g., "/tmp/hadoop-[username]/mapred/local/0_0/
+ * taskTracker/[username]/jobcache/job_xxx/attempt_xxx_r_xxx/output/"), and
+ * all sub-MapTasks create the same filename ("file.out"). Rename that to
+ * something unique (e.g., "map_0.out") to avoid collisions.
+ *
+ * Longer-term, we'll modify TaskTracker or whatever to use TaskAttemptID-
+ * based filenames instead of "file.out". (All of this is entirely internal,
+ * so there are no particular compatibility issues.)
+ */
private void renameMapOutputForReduce(TaskAttemptID mapId,
MapOutputFile subMapOutputFile)
throws IOException {
@@ -312,8 +323,10 @@ class UberTask extends Task {
reporter.progress();
}
- // every map will produce file.out (in the same dir), so rename as we go
- // (longer-term, will use TaskAttemptIDs as part of name => avoid rename)
+ // Every map will produce "file.out" in the same (local, not HDFS!) dir,
+ // so rename to "map_#.out" as we go. (Longer-term, will use
+ // TaskAttemptIDs as part of name => avoid rename.) Note that this has
+ // nothing to do with the _temporary/attempt_xxx _HDFS_ subdir above!
if (numReduceTasks > 0) {
renameMapOutputForReduce(mapIds[j], map.getMapOutputFile());
}
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestMapReduceLazyOutput.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestMapReduceLazyOutput.java?rev=1079252&r1=1079251&r2=1079252&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestMapReduceLazyOutput.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestMapReduceLazyOutput.java
Tue Mar 8 06:00:35 2011
@@ -88,6 +88,9 @@ public class TestMapReduceLazyOutput ext
private static void runTestLazyOutput(Configuration conf, Path output,
int numReducers, boolean createLazily)
throws Exception {
+ // ubertask has just one output file (part-r-00000) != number of maptasks
+ conf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, false);
+
Job job = Job.getInstance(conf, "Test-Lazy-Output");
FileInputFormat.setInputPaths(job, INPUT);
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestCopyFiles.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestCopyFiles.java?rev=1079252&r1=1079251&r2=1079252&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestCopyFiles.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestCopyFiles.java
Tue Mar 8 06:00:35 2011
@@ -46,6 +46,7 @@ import org.apache.hadoop.hdfs.MiniDFSClu
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobContext;
import org.apache.hadoop.mapred.MiniMRCluster;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.tools.DistCp;
@@ -751,6 +752,8 @@ public class TestCopyFiles extends TestC
}
Configuration job = mr.createJobConf();
job.setLong("distcp.bytes.per.map", totsize / 3);
+ // if uberized, get only 1 log file, not 4 or 5 as in assert below
+ job.setBoolean(JobContext.JOB_UBERTASK_ENABLE, false);
ToolRunner.run(new DistCp(job),
new String[] {"-m", "100",
"-log",
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHadoopArchives.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHadoopArchives.java?rev=1079252&r1=1079251&r2=1079252&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHadoopArchives.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHadoopArchives.java
Tue Mar 8 06:00:35 2011
@@ -157,7 +157,8 @@ public class TestHadoopArchives extends
System.setErr(out);
final String results;
try {
- assertEquals(0, shell.run(new String[]{"-lsr", dir}));
+ assertEquals("non-zero exit status", 0,
+ shell.run(new String[]{"-lsr", dir}));
results = bytes.toString();
} finally {
IOUtils.closeStream(out);
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHarFileSystem.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHarFileSystem.java?rev=1079252&r1=1079251&r2=1079252&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHarFileSystem.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/TestHarFileSystem.java
Tue Mar 8 06:00:35 2011
@@ -124,15 +124,15 @@ public class TestHarFileSystem extends T
byte[] b = new byte[4];
int readBytes = fin.read(b);
fin.close();
- assertTrue("strings are equal ", (b[0] == "a".getBytes()[0]));
+ assertTrue("harFilea strings are unequal.", (b[0] == "a".getBytes()[0]));
fin = harFs.open(harFileb);
fin.read(b);
fin.close();
- assertTrue("strings are equal ", (b[0] == "b".getBytes()[0]));
+ assertTrue("harFileb strings are unequal.", (b[0] == "b".getBytes()[0]));
fin = harFs.open(harFilec);
fin.read(b);
fin.close();
- assertTrue("strings are equal ", (b[0] == "c".getBytes()[0]));
+ assertTrue("harFilec strings are unequal.", (b[0] == "c".getBytes()[0]));
}
private void checkProperties(Path harPath, Configuration conf) throws
IOException {
@@ -165,7 +165,8 @@ public class TestHarFileSystem extends T
private void checkBlockSize(FileSystem fs, Path finalPath, long blockSize)
throws IOException {
FileStatus[] statuses = fs.globStatus(new Path(finalPath, "part-*"));
for (FileStatus status: statuses) {
- assertTrue(status.getBlockSize() == blockSize);
+ assertEquals("unexpected blocksize for part-* file", blockSize,
+ status.getBlockSize());
}
}
@@ -184,19 +185,20 @@ public class TestHarFileSystem extends T
args[4] = "test";
args[5] = archivePath.toString();
int ret = ToolRunner.run(har, args);
- assertTrue("failed test", ret == 0);
+ assertEquals("ToolRunner.run('-archiveName') failed.", 0, ret);
Path finalPath = new Path(archivePath, "foo1.har");
Path fsPath = new Path(inputPath.toUri().getPath());
Path filePath = new Path(finalPath, "test");
// make it a har path
Path harPath = new Path("har://" + filePath.toUri().getPath());
- assertTrue(fs.exists(new Path(finalPath, "_index")));
- assertTrue(fs.exists(new Path(finalPath, "_masterindex")));
- /*check for existence of only 1 part file, since part file size == 2GB */
- assertTrue(fs.exists(new Path(finalPath, "part-0")));
- assertTrue(!fs.exists(new Path(finalPath, "part-1")));
- assertTrue(!fs.exists(new Path(finalPath, "part-2")));
- assertTrue(!fs.exists(new Path(finalPath, "_logs")));
+ assertTrue("_index missing!", fs.exists(new Path(finalPath, "_index")));
+ assertTrue("_masterindex missing!",
+ fs.exists(new Path(finalPath, "_masterindex")));
+ // check for existence of only 1 part file, since part file size == 2GB
+ assertTrue("part-0 missing!", fs.exists(new Path(finalPath, "part-0")));
+ assertTrue("part-1 exists!", !fs.exists(new Path(finalPath, "part-1")));
+ assertTrue("part-2 exists!", !fs.exists(new Path(finalPath, "part-2")));
+ assertTrue("_logs exists!", !fs.exists(new Path(finalPath, "_logs")));
FileStatus[] statuses = fs.listStatus(finalPath);
args = new String[2];
args[0] = "-ls";
@@ -204,7 +206,7 @@ public class TestHarFileSystem extends T
FsShell shell = new FsShell(conf);
ret = ToolRunner.run(shell, args);
// fileb and filec
- assertTrue(ret == 0);
+ assertEquals("ToolRunner.run('-ls') failed.", 0, ret);
checkBytes(harPath, conf);
checkProperties(harPath, conf);
/* check block size for path files */
@@ -223,7 +225,7 @@ public class TestHarFileSystem extends T
args[6] = "test";
args[7] = archivePath.toString();
int ret = ToolRunner.run(har, args);
- assertTrue("failed test", ret == 0);
+ assertEquals("ToolRunner.run('-archiveName') failed.", 0, ret);
Path finalPath = new Path(archivePath, "foo.har");
Path fsPath = new Path(inputPath.toUri().getPath());
Path filePath = new Path(finalPath, "test");
@@ -243,7 +245,7 @@ public class TestHarFileSystem extends T
FsShell shell = new FsShell(conf);
ret = ToolRunner.run(shell, args);
// fileb and filec
- assertTrue(ret == 0);
+ assertEquals("ToolRunner.run('-ls') failed.", 0, ret);
checkBytes(harPath, conf);
checkProperties(harPath, conf);
checkBlockSize(fs, finalPath, 512);
@@ -256,13 +258,13 @@ public class TestHarFileSystem extends T
HadoopArchives har = new HadoopArchives(conf);
String[] args = new String[4];
- //check for destination not specfied
+ //check for destination not specified
args[0] = "-archiveName";
args[1] = "foo.har";
args[2] = "-p";
args[3] = "/";
int ret = ToolRunner.run(har, args);
- assertTrue(ret != 0);
+ assertFalse("bogus ToolRunner.run('-archiveName') didn't fail??", 0 ==
ret);
args = new String[6];
//check for wrong archiveName
args[0] = "-archiveName";
@@ -272,12 +274,12 @@ public class TestHarFileSystem extends T
args[4] = inputrelPath.toString();
args[5] = archivePath.toString();
ret = ToolRunner.run(har, args);
- assertTrue(ret != 0);
- // se if dest is a file
+ assertFalse("bogus ToolRunner.run('-archiveName') didn't fail??", 0 ==
ret);
+ // see if dest is a file
args[1] = "foo.har";
args[5] = filec.toString();
ret = ToolRunner.run(har, args);
- assertTrue(ret != 0);
+ assertFalse("bogus ToolRunner.run('-archiveName') didn't fail??", 0 ==
ret);
//this is a valid run
args[0] = "-archiveName";
args[1] = "foo.har";
@@ -286,12 +288,12 @@ public class TestHarFileSystem extends T
args[4] = inputrelPath.toString();
args[5] = archivePath.toString();
ret = ToolRunner.run(har, args);
- //checl for the existenece of the archive
- assertTrue(ret == 0);
+ // check for the existence of the archive
+ assertEquals("ToolRunner.run('-archiveName') failed.", 0, ret);
///try running it again. it should not
// override the directory
ret = ToolRunner.run(har, args);
- assertTrue(ret != 0);
+ assertTrue("ToolRunner.run('-archiveName') rerun did NOT fail.", ret != 0);
Path finalPath = new Path(archivePath, "foo.har");
Path fsPath = new Path(inputPath.toUri().getPath());
String relative = fsPath.toString().substring(1);
@@ -313,9 +315,8 @@ public class TestHarFileSystem extends T
args[1] = harPath.toString();
ret = ToolRunner.run(shell, args);
// ls should work.
- assertTrue((ret == 0));
- //now check for contents of filea
- // fileb and filec
+ assertEquals("ToolRunner.run('-ls') failed.", 0, ret);
+ // now check for contents of filea, fileb, and filec
Path harFilea = new Path(harPath, "a");
Path harFileb = new Path(harPath, "b");
Path harFilec = new Path(harPath, "c c");
@@ -325,17 +326,17 @@ public class TestHarFileSystem extends T
int readBytes = fin.read(b);
assertTrue("Empty read.", readBytes > 0);
fin.close();
- assertTrue("strings are equal ", (b[0] == "a".getBytes()[0]));
+ assertTrue("harFilea strings are unequal.", (b[0] == "a".getBytes()[0]));
fin = harFs.open(harFileb);
readBytes = fin.read(b);
assertTrue("Empty read.", readBytes > 0);
fin.close();
- assertTrue("strings are equal ", (b[0] == "b".getBytes()[0]));
+ assertTrue("harFileb strings are unequal.", (b[0] == "b".getBytes()[0]));
fin = harFs.open(harFilec);
readBytes = fin.read(b);
assertTrue("Empty read.", readBytes > 0);
fin.close();
- assertTrue("strings are equal ", (b[0] == "c".getBytes()[0]));
+ assertTrue("harFilec strings are unequal.", (b[0] == "c".getBytes()[0]));
// ok all files match
// run a map reduce job
FileSystem fsHar = harPath.getFileSystem(conf);
@@ -360,11 +361,12 @@ public class TestHarFileSystem extends T
FSDataInputStream reduceIn = fs.open(reduceFile);
b = new byte[6];
readBytes = reduceIn.read(b);
- assertTrue("Should read 6 bytes instead of "+readBytes+".", readBytes ==
6);
+ assertEquals("Should read 6 bytes instead of "+readBytes+".", 6,
readBytes);
//assuming all the 6 bytes were read.
Text readTxt = new Text(b);
- assertTrue("a\nb\nc\n".equals(readTxt.toString()));
- assertTrue("number of bytes left should be -1", reduceIn.read(b) == -1);
+ assertTrue("string read from reduceIn doesn't match expected value.",
+ "a\nb\nc\n".equals(readTxt.toString()));
+ assertEquals("number of bytes left should be -1.", -1, reduceIn.read(b));
reduceIn.close();
}
}
