[DO NOT MERGE] Trying cascading3

.gitignore

@@ -31,6 +31,10 @@ tutorial/data/rightDiff.tsv
 tutorial/data/tmp3.tsv
 tutorial/data/jsonoutput0.tsv
 tutorial/data/avrooutput0.avro
+tutorial/data/execution_output.txt
 .scalding_repl
 scalding-hadoop-test/NOTICE
 NOTICE
+*~
+build/
+.staging

.travis.yml

@@ -46,6 +46,22 @@ matrix:
       env: BUILD="base" TEST_TARGET="scalding-hadoop-test"
       script: "scripts/run_test.sh"
 
+    - scala: 2.10.6
+      env: BUILD="base" TEST_TARGET="scalding-hadoop2-mr1-test"
+      script: "scripts/run_test.sh"
+
+    - scala: 2.11.7
+      env: BUILD="base" TEST_TARGET="scalding-hadoop2-mr1-test"
+      script: "scripts/run_test.sh"
+
+    - scala: 2.10.6
+      env: BUILD="base" TEST_TARGET="scalding-hadoop2-tez-test"
+      script: "scripts/run_test.sh"
+
+    - scala: 2.11.7
+      env: BUILD="base" TEST_TARGET="scalding-hadoop2-tez-test"
+      script: "scripts/run_test.sh"
+
     - scala: 2.10.6
       env: BUILD="base" TEST_TARGET="scalding-serialization"
       script: "scripts/run_test.sh"

maple/src/main/java/com/etsy/cascading/tap/local/LocalTap.java

@@ -5,6 +5,7 @@
 import java.util.Properties;
 import java.util.logging.Logger;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.OutputCollector;
 import org.apache.hadoop.mapred.RecordReader;
@@ -45,18 +46,18 @@ public class LocalTap<SourceCtx, SinkCtx> extends Tap<Properties, RecordReader,
     private JobConf defaults;
     private Lfs lfs;
 
-    public LocalTap(String path, Scheme<JobConf, RecordReader, OutputCollector, SourceCtx, SinkCtx> scheme,
+    public LocalTap(String path, Scheme<Configuration, RecordReader, OutputCollector, SourceCtx, SinkCtx> scheme,
             SinkMode sinkMode) {
         super(new LocalScheme<SourceCtx, SinkCtx>(scheme), sinkMode);
         setup(path, scheme);
     }
 
-    public LocalTap(String path, Scheme<JobConf, RecordReader, OutputCollector, SourceCtx, SinkCtx> scheme) {
+    public LocalTap(String path, Scheme<Configuration, RecordReader, OutputCollector, SourceCtx, SinkCtx> scheme) {
         super(new LocalScheme<SourceCtx, SinkCtx>(scheme));
         setup(path, scheme);
     }
 
-    private void setup(String path, Scheme<JobConf, RecordReader, OutputCollector, SourceCtx, SinkCtx> scheme) {
+    private void setup(String path, Scheme<Configuration, RecordReader, OutputCollector, SourceCtx, SinkCtx> scheme) {
         this.path = path;
 
         /*
@@ -90,13 +91,13 @@ public String getIdentifier() {
     }
 
     @Override
-    public TupleEntryIterator openForRead(FlowProcess<Properties> flowProcess, RecordReader input) throws IOException {
+    public TupleEntryIterator openForRead(FlowProcess<? extends Properties> flowProcess, RecordReader input) throws IOException {
         JobConf jobConf = mergeDefaults("LocalTap#openForRead", flowProcess.getConfigCopy(), defaults);
         return lfs.openForRead(new HadoopFlowProcess(jobConf));
     }
 
     @Override
-    public TupleEntryCollector openForWrite(FlowProcess<Properties> flowProcess, OutputCollector output)
+    public TupleEntryCollector openForWrite(FlowProcess<? extends Properties> flowProcess, OutputCollector output)
             throws IOException {
         JobConf jobConf = mergeDefaults("LocalTap#openForWrite", flowProcess.getConfigCopy(), defaults);
         return lfs.openForWrite(new HadoopFlowProcess(jobConf));
@@ -141,11 +142,11 @@ private static class LocalScheme<SourceContext, SinkContext> extends
             Scheme<Properties, RecordReader, OutputCollector, SourceContext, SinkContext> {
         private static final long serialVersionUID = 5710119342340369543L;
 
-        private Scheme<JobConf, RecordReader, OutputCollector, SourceContext, SinkContext> scheme;
+        private Scheme<Configuration, RecordReader, OutputCollector, SourceContext, SinkContext> scheme;
         private JobConf defaults;
         private Lfs lfs;
 
-        public LocalScheme(Scheme<JobConf, RecordReader, OutputCollector, SourceContext, SinkContext> scheme) {
+        public LocalScheme(Scheme<Configuration, RecordReader, OutputCollector, SourceContext, SinkContext> scheme) {
             super(scheme.getSourceFields(), scheme.getSinkFields());
             this.scheme = scheme;
         }
@@ -159,53 +160,53 @@ private void setLfs(Lfs lfs) {
         }
 
         @Override
-        public Fields retrieveSourceFields(FlowProcess<Properties> flowProcess,
+        public Fields retrieveSourceFields(FlowProcess<? extends Properties> flowProcess,
                 Tap tap) {
             return scheme.retrieveSourceFields(new HadoopFlowProcess(defaults), lfs);
         }
 
         @Override
-        public void presentSourceFields(FlowProcess<Properties> flowProcess, 
+        public void presentSourceFields(FlowProcess<? extends Properties> flowProcess,
                 Tap tap, Fields fields) {
             scheme.presentSourceFields(new HadoopFlowProcess(defaults), lfs, fields);
         }
 
         @Override
-        public void sourceConfInit(FlowProcess<Properties> flowProcess,
+        public void sourceConfInit(FlowProcess<? extends Properties> flowProcess,
                 Tap<Properties, RecordReader, OutputCollector> tap, Properties conf) {
             JobConf jobConf = mergeDefaults("LocalScheme#sourceConfInit", conf, defaults);
             scheme.sourceConfInit(new HadoopFlowProcess(jobConf), lfs, jobConf);
             overwriteProperties(conf, jobConf);
         }
 
         @Override
-        public Fields retrieveSinkFields(FlowProcess<Properties> flowProcess,
+        public Fields retrieveSinkFields(FlowProcess<? extends Properties> flowProcess,
                 Tap tap) {
             return scheme.retrieveSinkFields(new HadoopFlowProcess(defaults), lfs);
         }
 
         @Override
-        public void presentSinkFields(FlowProcess<Properties> flowProcess, 
+        public void presentSinkFields(FlowProcess<? extends Properties> flowProcess,
                 Tap tap, Fields fields) {
             scheme.presentSinkFields(new HadoopFlowProcess(defaults), lfs, fields);
         }
 
         @Override
-        public void sinkConfInit(FlowProcess<Properties> flowProcess,
+        public void sinkConfInit(FlowProcess<? extends Properties> flowProcess,
                 Tap<Properties, RecordReader, OutputCollector> tap, Properties conf) {
             JobConf jobConf = mergeDefaults("LocalScheme#sinkConfInit", conf, defaults);
             scheme.sinkConfInit(new HadoopFlowProcess(jobConf), lfs, jobConf);
             overwriteProperties(conf, jobConf);
         }
 
         @Override
-        public boolean source(FlowProcess<Properties> flowProcess, SourceCall<SourceContext, RecordReader> sourceCall)
+        public boolean source(FlowProcess<? extends Properties> flowProcess, SourceCall<SourceContext, RecordReader> sourceCall)
                 throws IOException {
             throw new RuntimeException("LocalTap#source is never called");
         }
 
         @Override
-        public void sink(FlowProcess<Properties> flowProcess, SinkCall<SinkContext, OutputCollector> sinkCall)
+        public void sink(FlowProcess<? extends Properties> flowProcess, SinkCall<SinkContext, OutputCollector> sinkCall)
                 throws IOException {
             throw new RuntimeException("LocalTap#sink is never called");
         }

maple/src/main/java/com/twitter/maple/hbase/HBaseScheme.java

@@ -154,7 +154,7 @@ public String[] getFamilyNames() {
   }
 
   @Override
-  public void sourcePrepare(FlowProcess<JobConf> flowProcess,
+  public void sourcePrepare(FlowProcess<? extends JobConf> flowProcess,
       SourceCall<Object[], RecordReader> sourceCall) {
     Object[] pair =
         new Object[]{sourceCall.getInput().createKey(), sourceCall.getInput().createValue()};
@@ -163,13 +163,13 @@ public void sourcePrepare(FlowProcess<JobConf> flowProcess,
   }
 
   @Override
-  public void sourceCleanup(FlowProcess<JobConf> flowProcess,
+  public void sourceCleanup(FlowProcess<? extends JobConf> flowProcess,
       SourceCall<Object[], RecordReader> sourceCall) {
     sourceCall.setContext(null);
   }
 
   @Override
-  public boolean source(FlowProcess<JobConf> flowProcess,
+  public boolean source(FlowProcess<? extends JobConf> flowProcess,
       SourceCall<Object[], RecordReader> sourceCall) throws IOException {
     Tuple result = new Tuple();
 
@@ -206,7 +206,7 @@ public boolean source(FlowProcess<JobConf> flowProcess,
   }
 
   @Override
-  public void sink(FlowProcess<JobConf> flowProcess, SinkCall<Object[], OutputCollector> sinkCall)
+  public void sink(FlowProcess<? extends JobConf> flowProcess, SinkCall<Object[], OutputCollector> sinkCall)
       throws IOException {
     TupleEntry tupleEntry = sinkCall.getOutgoingEntry();
     OutputCollector outputCollector = sinkCall.getOutput();
@@ -231,7 +231,7 @@ public void sink(FlowProcess<JobConf> flowProcess, SinkCall<Object[], OutputColl
   }
 
   @Override
-  public void sinkConfInit(FlowProcess<JobConf> process,
+  public void sinkConfInit(FlowProcess<? extends JobConf> process,
       Tap<JobConf, RecordReader, OutputCollector> tap, JobConf conf) {
     conf.setOutputFormat(TableOutputFormat.class);
 
@@ -240,7 +240,7 @@ public void sinkConfInit(FlowProcess<JobConf> process,
   }
 
   @Override
-  public void sourceConfInit(FlowProcess<JobConf> process,
+  public void sourceConfInit(FlowProcess<? extends JobConf> process,
       Tap<JobConf, RecordReader, OutputCollector> tap, JobConf conf) {
     conf.setInputFormat(TableInputFormat.class);
 

maple/src/main/java/com/twitter/maple/hbase/HBaseTap.java

@@ -143,7 +143,7 @@ protected HBaseAdmin getHBaseAdmin(JobConf conf) throws MasterNotRunningExceptio
   }
 
   @Override
-  public void sinkConfInit(FlowProcess<JobConf> process, JobConf conf) {
+  public void sinkConfInit(FlowProcess<? extends JobConf> process, JobConf conf) {
     if(quorumNames != null) {
       conf.set("hbase.zookeeper.quorum", quorumNames);
     }
@@ -178,12 +178,12 @@ public String getIdentifier() {
   }
 
   @Override
-  public TupleEntryIterator openForRead(FlowProcess<JobConf> jobConfFlowProcess, RecordReader recordReader) throws IOException {
+  public TupleEntryIterator openForRead(FlowProcess<? extends JobConf> jobConfFlowProcess, RecordReader recordReader) throws IOException {
     return new HadoopTupleEntrySchemeIterator(jobConfFlowProcess, this, recordReader);
   }
 
   @Override
-  public TupleEntryCollector openForWrite(FlowProcess<JobConf> jobConfFlowProcess, OutputCollector outputCollector) throws IOException {
+  public TupleEntryCollector openForWrite(FlowProcess<? extends JobConf> jobConfFlowProcess, OutputCollector outputCollector) throws IOException {
     HBaseTapCollector hBaseCollector = new HBaseTapCollector( jobConfFlowProcess, this );
     hBaseCollector.prepare();
     return hBaseCollector;
@@ -230,7 +230,7 @@ public long getModifiedTime(JobConf jobConf) throws IOException {
   }
 
   @Override
-  public void sourceConfInit(FlowProcess<JobConf> process, JobConf conf) {
+  public void sourceConfInit(FlowProcess<? extends JobConf> process, JobConf conf) {
     // a hack for MultiInputFormat to see that there is a child format
     FileInputFormat.setInputPaths( conf, getPath() );
 

maple/src/main/java/com/twitter/maple/hbase/HBaseTapCollector.java

@@ -42,7 +42,7 @@ public class HBaseTapCollector extends TupleEntrySchemeCollector<JobConf, TupleE
   /** Field writer */
   private RecordWriter writer;
   /** Field flowProcess */
-  private final FlowProcess<JobConf> hadoopFlowProcess;
+  private final FlowProcess<? extends JobConf> hadoopFlowProcess;
   /** Field tap */
   private final Tap<JobConf, RecordReader, OutputCollector> tap;
   /** Field reporter */
@@ -58,7 +58,7 @@ public class HBaseTapCollector extends TupleEntrySchemeCollector<JobConf, TupleE
    * @throws IOException
    *           when fails to initialize
    */
-  public HBaseTapCollector(FlowProcess<JobConf> flowProcess, Tap<JobConf, RecordReader, OutputCollector> tap) throws IOException {
+  public HBaseTapCollector(FlowProcess<? extends JobConf> flowProcess, Tap<JobConf, RecordReader, OutputCollector> tap) throws IOException {
     super(flowProcess, tap.getScheme());
     this.hadoopFlowProcess = flowProcess;
     this.tap = tap;

maple/src/main/java/com/twitter/maple/tap/MemorySinkTap.java

@@ -7,6 +7,7 @@
 import cascading.tuple.Tuple;
 import cascading.tuple.TupleEntry;
 import cascading.tuple.TupleEntryIterator;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.JobConf;
 
 import java.io.File;
@@ -43,7 +44,7 @@ public static String getTempDir() {
     }
 
     @Override
-    public boolean commitResource(JobConf conf) throws java.io.IOException {
+    public boolean commitResource(Configuration conf) throws java.io.IOException {
         TupleEntryIterator it = new HadoopFlowProcess(conf).openTapForRead(this);
 
         boolean first_time = true;

maple/src/main/java/com/twitter/maple/tap/MemorySourceTap.java

@@ -46,21 +46,21 @@ public List<Tuple> getTuples() {
         }
 
         @Override
-        public void sourceConfInit(FlowProcess<JobConf> flowProcess,
+        public void sourceConfInit(FlowProcess<? extends JobConf> flowProcess,
             Tap<JobConf, RecordReader<TupleWrapper, NullWritable>, Void> tap, JobConf conf) {
             FileInputFormat.setInputPaths(conf, this.id);
             conf.setInputFormat(TupleMemoryInputFormat.class);
             TupleMemoryInputFormat.storeTuples(conf, TupleMemoryInputFormat.TUPLES_PROPERTY, this.tuples);
         }
 
         @Override
-        public void sinkConfInit(FlowProcess<JobConf> flowProcess,
+        public void sinkConfInit(FlowProcess<? extends JobConf> flowProcess,
             Tap<JobConf, RecordReader<TupleWrapper, NullWritable>, Void> tap, JobConf conf) {
             throw new UnsupportedOperationException("Not supported yet.");
         }
 
         @Override
-        public void sourcePrepare( FlowProcess<JobConf> flowProcess, SourceCall<Object[],
+        public void sourcePrepare( FlowProcess<? extends JobConf> flowProcess, SourceCall<Object[],
             RecordReader<TupleWrapper, NullWritable>> sourceCall ) {
             sourceCall.setContext( new Object[ 2 ] );
 
@@ -69,7 +69,7 @@ public void sourcePrepare( FlowProcess<JobConf> flowProcess, SourceCall<Object[]
         }
 
         @Override
-        public boolean source(FlowProcess<JobConf> flowProcess, SourceCall<Object[],
+        public boolean source(FlowProcess<? extends JobConf> flowProcess, SourceCall<Object[],
             RecordReader<TupleWrapper, NullWritable>> sourceCall) throws IOException {
             TupleWrapper key = (TupleWrapper) sourceCall.getContext()[ 0 ];
             NullWritable value = (NullWritable) sourceCall.getContext()[ 1 ];
@@ -84,13 +84,13 @@ public boolean source(FlowProcess<JobConf> flowProcess, SourceCall<Object[],
         }
 
         @Override
-        public void sourceCleanup( FlowProcess<JobConf> flowProcess, SourceCall<Object[],
+        public void sourceCleanup( FlowProcess<? extends JobConf> flowProcess, SourceCall<Object[],
             RecordReader<TupleWrapper, NullWritable>> sourceCall ) {
             sourceCall.setContext( null );
         }
 
         @Override
-        public void sink(FlowProcess<JobConf> flowProcess, SinkCall<Void, Void> sinkCall ) throws IOException {
+        public void sink(FlowProcess<? extends JobConf> flowProcess, SinkCall<Void, Void> sinkCall ) throws IOException {
             throw new UnsupportedOperationException("Not supported.");
         }
 
@@ -127,7 +127,7 @@ public boolean equals(Object object) {
     }
 
     @Override
-    public TupleEntryIterator openForRead( FlowProcess<JobConf> flowProcess, RecordReader<TupleWrapper,
+    public TupleEntryIterator openForRead( FlowProcess<? extends JobConf> flowProcess, RecordReader<TupleWrapper,
         NullWritable> input ) throws IOException {
         // input may be null when this method is called on the client side or cluster side when accumulating
         // for a HashJoin

maple/src/main/java/com/twitter/maple/tap/StdoutTap.java

@@ -5,6 +5,7 @@
 import cascading.tap.hadoop.Lfs;
 import cascading.tuple.Fields;
 import cascading.tuple.TupleEntryIterator;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.JobConf;
 
 import java.io.File;
@@ -32,7 +33,7 @@ public static String getTempDir() {
     }
 
     @Override
-    public boolean commitResource(JobConf conf) throws java.io.IOException {
+    public boolean commitResource(Configuration conf) throws java.io.IOException {
         TupleEntryIterator it = new HadoopFlowProcess(conf).openTapForRead(this);
         System.out.println("");
         System.out.println("");