Input/Output format classes are now supported for map reduce jobs

Author: amir343

README.md

@@ -11,6 +11,9 @@ The classic word counts example looks like this in Saaloop:
 
 ```scala
 import com.jayway.saaloop.dsl.Saaloop._
+import org.apache.hadoop.io.{IntWritable, Text, LongWritable}
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat
 
 object WordCount {
 
@@ -31,9 +34,12 @@ object WordCount {
     }
 
     job("myjob")(
+      hadoopConfiguration = c,
       mapWith = m,
       reduceWith = r,
-      hadoopConfiguration = c
+      inputFormatClass = classOf[TextInputFormat],
+      outputFormatClass = classOf[TextOutputFormat[_,_]],
+      waitForCompletion = true
     )
   }
 

saaloop-core/src/main/scala/com/jayway/saaloop/dsl/SaaloopJob.scala

@@ -2,6 +2,9 @@ package com.jayway.saaloop.dsl
 
 import org.apache.hadoop.mapreduce.{Reducer, Job, Mapper}
 import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat
+import org.apache.hadoop.fs.Path
 
 /**
  * Copyright 2012 Amir Moulavi ([email protected])
@@ -27,8 +30,15 @@ trait SaaloopJob extends Types {
 
   object job {
 
-    def apply[K1 <: key, V1 <: value, K2 <: key, V2 <: value, K3 <: key, V3 <: value, K4 <: key, V4 <: value]
-      (name:String)(mapWith:Mapper[K1, V1, K2, V2], reduceWith:Reducer[K3, V3, K4, V4], hadoopConfiguration:Configuration = conf):Job = {
+    def apply[K1 <: key, V1 <: value, K2 <: key, V2 <: value, K3 <: key, V3 <: value, K4 <: key, V4 <: value, FIN <: fin, FOUT <: fout]
+      (name:String)(mapWith:Mapper[K1, V1, K2, V2], 
+                    reduceWith:Reducer[K3, V3, K4, V4], 
+                    hadoopConfiguration:Configuration = conf,
+                    inputPath:String = ".",
+                    outputPath:String = ".",
+                    inputFormatClass:Class[FIN] = null,
+                    outputFormatClass:Class[FOUT] = null,
+                    waitForCompletion:Boolean = false):Job = {
 
       val job = new Job(hadoopConfiguration, name)
       job.setMapperClass(mapWith.getClass)
@@ -37,6 +47,11 @@ trait SaaloopJob extends Types {
       val outputValueType = asInstanceOf[V4]
       job.setOutputKeyClass(outputKeyType.getClass)
       job.setOutputValueClass(outputValueType.getClass)
+      FileInputFormat.addInputPath(job, new Path(inputPath))
+      FileOutputFormat.setOutputPath(job, new Path(outputPath))
+      if (inputFormatClass != null) job.setInputFormatClass(inputFormatClass)
+      if (outputFormatClass != null) job.setOutputFormatClass(outputFormatClass)
+      job.waitForCompletion(waitForCompletion)
 
       job
     }

saaloop-core/src/main/scala/com/jayway/saaloop/dsl/Types.scala

@@ -1,6 +1,8 @@
 package com.jayway.saaloop.dsl
 
 import org.apache.hadoop.io.{WritableComparable, Writable}
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat
 
 /**
  * Copyright 2012 Amir Moulavi ([email protected])
@@ -23,4 +25,6 @@ import org.apache.hadoop.io.{WritableComparable, Writable}
 trait Types {
   type key = Writable with WritableComparable[_]
   type value = Writable
+  type fin = FileInputFormat[_, _]
+  type fout = FileOutputFormat[_, _]
 }

saaloop-examples/src/main/scala/WordCount.scala

@@ -1,5 +1,7 @@
 import com.jayway.saaloop.dsl.Saaloop._
 import org.apache.hadoop.io.{IntWritable, Text, LongWritable}
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat
 
 /**
  * Copyright 2012 Amir Moulavi ([email protected])
@@ -38,9 +40,12 @@ object WordCount {
     }
 
     job("myjob")(
+      hadoopConfiguration = c,
       mapWith = m,
       reduceWith = r,
-      hadoopConfiguration = c
+      inputFormatClass = classOf[TextInputFormat],
+      outputFormatClass = classOf[TextOutputFormat[_,_]],
+      waitForCompletion = true
     )
   }