Add support for filecrushing on Elastic MapReduce

README

@@ -159,8 +159,8 @@ Now we try an example using the directory options. Say we invoke the crush like
   Crush \
   --regex=.*/(.+) \
   --replacement=$1-${crush.timestamp}-${crush.task.num}-${crush.file.num} \
-  --input=sequence \
-  --output=sequence \
+  --input-format=sequence \
+  --output-format=sequence \
   /user/example/work/input /user/example/work/output 20100221175612
 
 The --regex and --replacement arguments are similar to the arguments passed to String.replaceAll(). The regex argument matches the final part of a directory path. For /user/example/work/input, it will match input. For /user/example/work/input/subdir, it will match subdir. For matching purposes, a directory path does not have a trailing slash. The replacement argument refers to the match group by number to rename the file. The result is:
@@ -179,21 +179,21 @@ The following invocation fails:
   Crush \
   --regex=.*/input \
   --replacement=input-${crush.timestamp}-${crush.task.num}-${crush.file.num} \
-  --input=sequence \ 
-  --output=sequence \
+  --input-format=sequence \ 
+  --output-format=sequence \
   /user/example/work/input /user/example/work/output 20100221175612
 
 Since we have specified some directory options, we must ensure that all directories in hierarchy rooted at the input argument have a matching regex (since the default regex is no longer applicable). In this invocation, there is no regex argument that matches /user/example/work/input/subdir. We must change it to:
 
   Crush \
   --regex=.*/input \
   --replacement=input-${crush.timestamp}-${crush.task.num}-${crush.file.num} \
-  --input=sequence \
-  --output=sequence \
+  --input-format=sequence \
+  --output-format=sequence \
   --regex=.*/subdir \
   --replacement=as-text-${crush.timestamp}-${crush.task.num}-${crush.file.num} \
-  --input=sequence \
-  --output=text \
+  --input-format=sequence \
+  --output-format=text \
   /user/example/work/input /user/example/work/output 20100221175612
 
 This will yield:

pom.xml

@@ -4,7 +4,7 @@
 	<groupId>com.m6d</groupId>
 	<artifactId>filecrush</artifactId>
 	<name>M6D App - Filecrush</name>
-	<version>2.2.2-SNAPSHOT</version>
+	<version>2.2.3-SNAPSHOT</version>
 	<description>filecrush utility</description>
 	<packaging>jar</packaging>
 	<properties>

src/main/java/com/m6d/filecrush/clean/Clean.java

@@ -51,8 +51,10 @@ public static void main(String[] args) throws Exception {
 	public int run(String[] args) throws Exception {
         conf = getConf();
 
+        Path targetDir = new Path(conf.get(TARGET_DIR));
+
 		try {
-			fs=FileSystem.get(getConf());
+			fs = targetDir.getFileSystem(conf);
 		} catch (IOException e) {
 			throw new RuntimeException("Could not open filesystem");
 		}
@@ -67,7 +69,7 @@ public int run(String[] args) throws Exception {
 			cutoff=now-targetAge;
 		}
 
-        return cleanup (new Path(conf.get(TARGET_DIR)));
+        return cleanup(targetDir);
 
 	}
 

src/main/java/com/m6d/filecrush/crush/CountersInputFormat.java

@@ -50,7 +50,7 @@ public RecordReader<Counters, NullWritable> getRecordReader(InputSplit inputSpli
 		Path path = fSplit.getPath();
 		long length = fSplit.getLength();
 
-		FileSystem fs = FileSystem.get(jobconf);
+		FileSystem fs = path.getFileSystem(jobconf);
 
 		FSDataInputStream is = fs.open(path);
 

src/main/java/com/m6d/filecrush/crush/Crush.java

@@ -574,7 +574,7 @@ public int run(String[] args) throws Exception {
 			return 0;
 		}
 
-		setFileSystem(FileSystem.get(job));
+		setFileSystem(srcDir.getFileSystem(job));
 
 		FileStatus status = fs.getFileStatus(srcDir);
 

src/main/java/com/m6d/filecrush/crush/CrushPartitioner.java

@@ -41,9 +41,10 @@ public void configure(JobConf job) {
 		bucketToPartition = new HashMap<Text, Integer>(100);
 
 		try {
-			FileSystem fs = FileSystem.get(job);
+			Path p = new Path(path);
+			FileSystem fs = p.getFileSystem(job);
 
-			Reader reader = new Reader(fs, new Path(path), job);
+			Reader reader = new Reader(fs, p, job);
 
 			Text bucket = new Text();
 			IntWritable partNum = new IntWritable();

src/main/java/com/m6d/filecrush/crush/CrushReducer.java

@@ -127,7 +127,8 @@ public void configure(JobConf job) {
 		 * The files we write should be rooted in the "crush" subdir of the output directory to distinguish them from the files
 		 * created by the collector.
 		 */
-		outDirPath = new Path(outDirPath + "/crush").toUri().getPath();
+		Path outDirP = new Path(outDirPath + "/crush");
+		outDirPath = outDirP.toUri().getPath();
 
 		/*
 		 * Configure the regular expressions and replacements we use to convert dir names to crush output file names. Also get the
@@ -145,7 +146,7 @@ public void configure(JobConf job) {
 		placeHolderToValue.put("crush.timestamp", job.get("crush.timestamp"));
 
 		try {
-			fs = FileSystem.get(job);
+			fs = outDirP.getFileSystem(job);
 		} catch (RuntimeException e) {
 			throw e;
 		} catch (Exception e) {