diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/AbstractHadoopOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/AbstractHadoopOperatorDescriptor.java
index 9b054e4..2d7bc84 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/AbstractHadoopOperatorDescriptor.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/AbstractHadoopOperatorDescriptor.java
@@ -14,18 +14,10 @@
  */
 package edu.uci.ics.hyracks.dataflow.hadoop;
 
-import java.io.File;
 import java.io.IOException;
-import java.lang.reflect.Method;
-import java.net.URL;
-import java.net.URLClassLoader;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.StringTokenizer;
 
-import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.InputSplit;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.OutputCollector;
@@ -42,8 +34,9 @@
 
 public abstract class AbstractHadoopOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
 
-	
-	protected static class DataWritingOutputCollector<K, V> implements OutputCollector<K, V> {
+    protected transient JobConf jobConf;
+
+    protected static class DataWritingOutputCollector<K, V> implements OutputCollector<K, V> {
         private IDataWriter<Object[]> writer;
 
         public DataWritingOutputCollector() {
@@ -55,7 +48,7 @@
 
         @Override
         public void collect(Object key, Object value) throws IOException {
-           	writer.writeData(new Object[] { key, value });
+            writer.writeData(new Object[] { key, value });
         }
 
         public void setWriter(IDataWriter<Object[]> writer) {
@@ -69,38 +62,28 @@
     private static final long serialVersionUID = 1L;
     private final Map<String, String> jobConfMap;
     private IHadoopClassFactory hadoopClassFactory;
-    
-    public abstract RecordDescriptor getRecordDescriptor(JobConf jobConf);
-    	
+
     public AbstractHadoopOperatorDescriptor(JobSpecification spec, RecordDescriptor recordDescriptor, JobConf jobConf,
             IHadoopClassFactory hadoopOperatorFactory) {
         super(spec, 1, 1);
         jobConfMap = DatatypeHelper.jobConf2Map(jobConf);
         this.hadoopClassFactory = hadoopOperatorFactory;
-        if(recordDescriptor != null){
-        	recordDescriptors[0]= recordDescriptor;
-        }else{
-        	recordDescriptors[0] = getRecordDescriptor(jobConf);
-        }
+        recordDescriptors[0] = recordDescriptor;
     }
 
-     
     public Map<String, String> getJobConfMap() {
-		return jobConfMap;
-	}
+        return jobConfMap;
+    }
 
+    public IHadoopClassFactory getHadoopClassFactory() {
+        return hadoopClassFactory;
+    }
 
-	public IHadoopClassFactory getHadoopClassFactory() {
-		return hadoopClassFactory;
-	}
+    public void setHadoopClassFactory(IHadoopClassFactory hadoopClassFactory) {
+        this.hadoopClassFactory = hadoopClassFactory;
+    }
 
-
-	public void setHadoopClassFactory(IHadoopClassFactory hadoopClassFactory) {
-		this.hadoopClassFactory = hadoopClassFactory;
-	}
-
-
-	protected Reporter createReporter() {
+    protected Reporter createReporter() {
         return new Reporter() {
             @Override
             public Counter getCounter(Enum<?> name) {
@@ -110,8 +93,7 @@
             @Override
             public Counter getCounter(String group, String name) {
                 return null;
-            }    
-              
+            }
 
             @Override
             public InputSplit getInputSplit() throws UnsupportedOperationException {
@@ -141,9 +123,13 @@
     }
 
     public JobConf getJobConf() {
-        return DatatypeHelper.map2JobConf(jobConfMap);
+        if (jobConf == null) {
+            jobConf = DatatypeHelper.map2JobConf(jobConfMap);
+            jobConf.setClassLoader(this.getClass().getClassLoader());
+        }
+        return jobConf;
     }
-    
+
     public void populateCache(JobConf jobConf) {
         String cache = jobConf.get(MAPRED_CACHE_FILES);
         System.out.println("cache:" + cache);
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HDFSWriteOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HDFSWriteOperatorDescriptor.java
deleted file mode 100644
index daaa5ea..0000000
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HDFSWriteOperatorDescriptor.java
+++ /dev/null
@@ -1,197 +0,0 @@
-/*
- * Copyright 2009-2010 by The Regents of the University of California
- * Licensed 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 from
- * 
- *     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 edu.uci.ics.hyracks.dataflow.hadoop;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.io.SequenceFile;
-import org.apache.hadoop.io.SequenceFile.Writer;
-import org.apache.hadoop.io.compress.CompressionCodec;
-import org.apache.hadoop.io.compress.GzipCodec;
-import org.apache.hadoop.mapred.FileOutputFormat;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.SequenceFileOutputFormat;
-
-import edu.uci.ics.hyracks.api.job.JobSpecification;
-import edu.uci.ics.hyracks.dataflow.hadoop.util.DatatypeHelper;
-import edu.uci.ics.hyracks.dataflow.std.file.AbstractFileWriteOperatorDescriptor;
-import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
-import edu.uci.ics.hyracks.dataflow.std.file.IRecordWriter;
-import edu.uci.ics.hyracks.dataflow.std.file.RecordWriter;
-
-
-
-public class HDFSWriteOperatorDescriptor extends
-		AbstractFileWriteOperatorDescriptor {
-	
-    private static String nullWritableClassName = NullWritable.class.getName();
-	
-    private static class HDFSWriter extends RecordWriter {
-		
-       HDFSWriter(FileSystem fileSystem, String hdfsPath, int[] columns, char separator)
-           throws Exception {
-           super(columns,separator,new Object[]{fileSystem,hdfsPath});
-       }       
-    
-
-       @Override
-       public OutputStream createOutputStream(Object[] args) throws Exception {
-	       FSDataOutputStream fs = ((FileSystem)args[0]).create(new Path((String)args[1]));
-	       return fs;
-       }
-
-       @Override
-       public void write(Object[] record) throws Exception {
-          if(!nullWritableClassName.equals((record[0].getClass().getName()))){
-              bufferedWriter.write(String.valueOf(record[0]));
-          }
-          if(!nullWritableClassName.equals((record[1].getClass().getName()))){
-              bufferedWriter.write(separator);	 
-              bufferedWriter.write(String.valueOf(record[1]));
-          }	 
-          bufferedWriter.write("\n");
-       }
-    }
-
-    private static class HDFSSequenceWriter extends RecordWriter {
-	private Writer writer;
-	HDFSSequenceWriter(FileSystem fileSystem, String hdfsPath, Writer writer)
-            throws Exception {
-            super(null,COMMA,new Object[]{fileSystem,hdfsPath});
-    	    this.writer = writer;
-        }
-
-	@Override
-	public OutputStream createOutputStream(Object[] args) throws Exception {
-		return null;
-	}
-		
-	@Override
-        public void close() {
-            try {
-                writer.close();
-            } catch (IOException e) {
-	          e.printStackTrace();
-	    }
-        }
-
-        @Override
-        public void write(Object[] record) throws Exception {
-             Object key = record[0];
-             Object value = record[1];
-             writer.append(key, value);
-        }
-    }
-	
-    private static final long serialVersionUID = 1L;
-    private static final char COMMA = ',';
-    private char separator;
-    private boolean sequenceFileOutput = false;
-    Map<String,String> jobConfMap;
-    
-
-    @Override
-    protected IRecordWriter createRecordWriter(File file,int index) throws Exception {
-    	JobConf conf = DatatypeHelper.map2JobConf((HashMap)jobConfMap);
-	System.out.println("replication:" + conf.get("dfs.replication"));
-    	FileSystem fileSystem = null;
-     	try {
-	    fileSystem = FileSystem.get(conf);
-	} catch (IOException ioe) {
-            ioe.printStackTrace();
-	}
-	Path path = new Path(file.getAbsolutePath());
-	checkIfCanWriteToHDFS(new FileSplit[]{new FileSplit("localhost",file)});
-	FSDataOutputStream outputStream = fileSystem.create(path);
-	outputStream.close();
-	if(sequenceFileOutput) {
-            Class keyClass = Class.forName(conf.getOutputKeyClass().getName());  
-	    Class valueClass= Class.forName(conf.getOutputValueClass().getName());
-            conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);
-   	    Writer writer = SequenceFile.createWriter(fileSystem, conf,path, keyClass, valueClass);
-	    return new HDFSSequenceWriter(fileSystem, file.getAbsolutePath(), writer);
-        } else {
-	    return new HDFSWriter(fileSystem, file.getAbsolutePath(), null, COMMA);
-        }	
-    }
-    
-    private boolean checkIfCanWriteToHDFS(FileSplit[] fileSplits) throws Exception{
-    	boolean canWrite = true;
-    	JobConf conf = DatatypeHelper.map2JobConf((HashMap)jobConfMap);
-	FileSystem fileSystem = null;
-	try {
-	    fileSystem = FileSystem.get(conf);
-	    for(FileSplit fileSplit : fileSplits) {
-		Path path = new Path(fileSplit.getLocalFile().getAbsolutePath());
-		canWrite = !fileSystem.exists(path);
-		if(!canWrite){
-	            throw new Exception(" Output path :  already exists");
-		}	
-            }
-	} catch(IOException ioe) {
-	    ioe.printStackTrace();
-	    throw ioe;
-	}
-        return canWrite;
-   }    
-
-    private static String getAbsolutePath(Path path) {
-          StringBuffer absolutePath = new StringBuffer();
-          List<String> ancestorPath = new ArrayList<String>();
-          Path pathElement=path;
-          while(pathElement != null) {
-                ancestorPath.add(0, pathElement.getName());
-             pathElement = pathElement.getParent();
-          }
-          ancestorPath.remove(0);        
-          for(String s : ancestorPath) {
-              absolutePath.append("/");
-              absolutePath.append(s);
-          }
-          return new String(absolutePath);
-     }
-
-     private static FileSplit[] getOutputSplits(JobConf conf,int noOfMappers){
-         int numOutputters = conf.getNumReduceTasks() != 0 ? conf.getNumReduceTasks() : noOfMappers;
-         FileSplit[] outputFileSplits = new FileSplit[numOutputters];
-         String absolutePath = getAbsolutePath(FileOutputFormat.getOutputPath(conf));
-         for(int index=0;index<numOutputters;index++) {
-             String suffix = new String("part-00000");
-             suffix = new String(suffix.substring(0, suffix.length() - ("" + index).length()));
-             suffix = suffix + index;
-             String outputPath = absolutePath + "/" + suffix;
-             outputFileSplits[index] = new FileSplit("localhost",new File(outputPath));
-         }
-         return outputFileSplits;
-    }
-
-    public HDFSWriteOperatorDescriptor(JobSpecification jobSpec,JobConf jobConf, int numMapTasks) throws Exception{
-        super(jobSpec,getOutputSplits(jobConf,numMapTasks));
-        this.jobConfMap = DatatypeHelper.jobConf2Map(jobConf);
-        checkIfCanWriteToHDFS(super.splits);
-	    this.sequenceFileOutput = 
-	    (jobConf.getOutputFormat() instanceof SequenceFileOutputFormat);
-    }
-}
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopMapperOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopMapperOperatorDescriptor.java
index 8397d86..6ebe0b9 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopMapperOperatorDescriptor.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopMapperOperatorDescriptor.java
@@ -17,6 +17,9 @@
 import java.io.IOException;
 
 import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapred.FileSplit;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.InputSplit;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.Mapper;
 import org.apache.hadoop.mapred.OutputCollector;
@@ -32,6 +35,7 @@
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.hadoop.util.DatatypeHelper;
 import edu.uci.ics.hyracks.dataflow.hadoop.util.IHadoopClassFactory;
+import edu.uci.ics.hyracks.dataflow.hadoop.util.InputSplitsProxy;
 import edu.uci.ics.hyracks.dataflow.std.base.IOpenableDataWriterOperator;
 import edu.uci.ics.hyracks.dataflow.std.util.DeserializedOperatorNodePushable;
 
@@ -41,6 +45,11 @@
         private Reporter reporter;
         private Mapper<K1, V1, K2, V2> mapper;
         private IOpenableDataWriter<Object[]> writer;
+        private int partition;
+
+        public MapperOperator(int partition) {
+            this.partition = partition;
+        };
 
         @Override
         public void close() throws HyracksDataException {
@@ -54,20 +63,39 @@
 
         @Override
         public void open() throws HyracksDataException {
-            JobConf jobConf = getJobConf();
+            jobConf = getJobConf();
             populateCache(jobConf);
             try {
                 mapper = createMapper();
             } catch (Exception e) {
                 throw new HyracksDataException(e);
             }
-            // -- - configure - --
+            if (inputSplitsProxy != null) {
+                updateConfWithSplit();
+            }
             mapper.configure(jobConf);
             writer.open();
             output = new DataWritingOutputCollector<K2, V2>(writer);
             reporter = createReporter();
         }
 
+        private void updateConfWithSplit() {
+            try {
+                InputSplit[] splits = inputSplitsProxy.toInputSplits(jobConf);
+                InputSplit splitRead = splits[partition];
+                if (splitRead instanceof FileSplit) {
+                    jobConf.set("map.input.file", ((FileSplit) splitRead).getPath().toString());
+                    jobConf.setLong("map.input.start", ((FileSplit) splitRead).getStart());
+                    jobConf.setLong("map.input.length", ((FileSplit) splitRead).getLength());
+                }
+            } catch (Exception e) {
+                e.printStackTrace();
+                // we do not throw the exception here as we are setting additional parameters that may not be 
+                // required by the mapper. If they are  indeed required,  the configure method invoked on the mapper
+                // shall report an exception because of the missing parameters. 
+            }
+        }
+
         @Override
         public void setDataWriter(int index, IOpenableDataWriter<Object[]> writer) {
             if (index != 0) {
@@ -87,32 +115,36 @@
     }
 
     private static final long serialVersionUID = 1L;
-    private static final String mapClassNameKey = "mapred.mapper.class";
     private Class<? extends Mapper> mapperClass;
+    private InputSplitsProxy inputSplitsProxy;
 
-    public HadoopMapperOperatorDescriptor(JobSpecification spec, Class<? extends Mapper> mapperClass,
-            RecordDescriptor recordDescriptor, JobConf jobConf) {
-        super(spec, recordDescriptor, jobConf, null);
-        this.mapperClass = mapperClass;
+    private void initializeSplitInfo(InputSplit[] splits) throws IOException {
+        jobConf = super.getJobConf();
+        InputFormat inputFormat = jobConf.getInputFormat();
+        inputSplitsProxy = new InputSplitsProxy(splits);
     }
 
-    public HadoopMapperOperatorDescriptor(JobSpecification spec, JobConf jobConf, IHadoopClassFactory hadoopClassFactory) {
-        super(spec, null, jobConf, hadoopClassFactory);
+    public HadoopMapperOperatorDescriptor(JobSpecification spec, JobConf jobConf, InputSplit[] splits,
+            IHadoopClassFactory hadoopClassFactory) throws IOException {
+        super(spec, getRecordDescriptor(jobConf, hadoopClassFactory), jobConf, hadoopClassFactory);
+        if (splits != null) {
+            initializeSplitInfo(splits);
+        }
     }
 
-    public RecordDescriptor getRecordDescriptor(JobConf conf) {
+    public static RecordDescriptor getRecordDescriptor(JobConf conf, IHadoopClassFactory hadoopClassFactory) {
         RecordDescriptor recordDescriptor = null;
         String mapOutputKeyClassName = conf.getMapOutputKeyClass().getName();
         String mapOutputValueClassName = conf.getMapOutputValueClass().getName();
         try {
-            if (getHadoopClassFactory() == null) {
-                recordDescriptor = DatatypeHelper.createKeyValueRecordDescriptor(
-                        (Class<? extends Writable>) Class.forName(mapOutputKeyClassName),
-                        (Class<? extends Writable>) Class.forName(mapOutputValueClassName));
+            if (hadoopClassFactory == null) {
+                recordDescriptor = DatatypeHelper.createKeyValueRecordDescriptor((Class<? extends Writable>) Class
+                        .forName(mapOutputKeyClassName), (Class<? extends Writable>) Class
+                        .forName(mapOutputValueClassName));
             } else {
                 recordDescriptor = DatatypeHelper.createKeyValueRecordDescriptor(
-                        (Class<? extends Writable>) getHadoopClassFactory().loadClass(mapOutputKeyClassName),
-                        (Class<? extends Writable>) getHadoopClassFactory().loadClass(mapOutputValueClassName));
+                        (Class<? extends Writable>) hadoopClassFactory.loadClass(mapOutputKeyClassName),
+                        (Class<? extends Writable>) hadoopClassFactory.loadClass(mapOutputValueClassName));
             }
         } catch (Exception e) {
             e.printStackTrace();
@@ -124,7 +156,7 @@
         if (mapperClass != null) {
             return mapperClass.newInstance();
         } else {
-            String mapperClassName = super.getJobConfMap().get(mapClassNameKey);
+            String mapperClassName = super.getJobConf().getMapperClass().getName();
             Object mapper = getHadoopClassFactory().createMapper(mapperClassName);
             mapperClass = (Class<? extends Mapper>) mapper.getClass();
             return (Mapper) mapper;
@@ -134,8 +166,8 @@
     @Override
     public IOperatorNodePushable createPushRuntime(IHyracksContext ctx, IOperatorEnvironment env,
             IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
-        return new DeserializedOperatorNodePushable(ctx, new MapperOperator(),
-                recordDescProvider.getInputRecordDescriptor(getOperatorId(), 0));
+        return new DeserializedOperatorNodePushable(ctx, new MapperOperator(partition), recordDescProvider
+                .getInputRecordDescriptor(getOperatorId(), 0));
     }
 
     public Class<? extends Mapper> getMapperClass() {
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReadOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReadOperatorDescriptor.java
index 53d1a7f..871ea25 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReadOperatorDescriptor.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReadOperatorDescriptor.java
@@ -20,13 +20,13 @@
 import java.util.Map;
 
 import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapred.Counters.Counter;
 import org.apache.hadoop.mapred.InputFormat;
 import org.apache.hadoop.mapred.InputSplit;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapred.SequenceFileRecordReader;
+import org.apache.hadoop.mapred.Counters.Counter;
 import org.apache.hadoop.util.ReflectionUtils;
 
 import edu.uci.ics.hyracks.api.constraints.PartitionCountConstraint;
@@ -52,11 +52,10 @@
     private Map<String, String> jobConfMap;
     private InputSplitsProxy inputSplitsProxy;
 
-    public HadoopReadOperatorDescriptor(JobConf jobConf, JobSpecification spec) throws IOException {
+    public HadoopReadOperatorDescriptor(JobConf jobConf, JobSpecification spec, InputSplit[] splits) throws IOException {
         super(spec, 0, 1);
         this.jobConfMap = DatatypeHelper.jobConf2Map(jobConf);
         InputFormat inputFormat = jobConf.getInputFormat();
-        InputSplit[] splits = inputFormat.getSplits(jobConf, jobConf.getNumMapTasks());
         RecordReader recordReader = inputFormat.getRecordReader(splits[0], jobConf, createReporter());
         recordDescriptors[0] = DatatypeHelper.createKeyValueRecordDescriptor((Class<? extends Writable>) recordReader
                 .createKey().getClass(), (Class<? extends Writable>) recordReader.createValue().getClass());
@@ -114,6 +113,7 @@
             public void initialize() throws HyracksDataException {
                 try {
                     JobConf conf = DatatypeHelper.map2JobConf((HashMap) jobConfMap);
+                    conf.setClassLoader(this.getClass().getClassLoader());
                     RecordReader hadoopRecordReader;
                     Writable key;
                     Writable value;
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReducerOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReducerOperatorDescriptor.java
index 2a70577..8a22641 100644
--- a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReducerOperatorDescriptor.java
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopReducerOperatorDescriptor.java
@@ -22,11 +22,11 @@
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.io.WritableComparator;
-import org.apache.hadoop.mapred.Counters.Counter;
 import org.apache.hadoop.mapred.InputSplit;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.Reducer;
 import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.Counters.Counter;
 
 import edu.uci.ics.hyracks.api.context.IHyracksContext;
 import edu.uci.ics.hyracks.api.dataflow.IDataReader;
@@ -41,12 +41,11 @@
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.hadoop.data.KeyComparatorFactory;
 import edu.uci.ics.hyracks.dataflow.hadoop.data.WritableComparingComparatorFactory;
-import edu.uci.ics.hyracks.dataflow.hadoop.util.ClasspathBasedHadoopClassFactory;
 import edu.uci.ics.hyracks.dataflow.hadoop.util.DatatypeHelper;
 import edu.uci.ics.hyracks.dataflow.hadoop.util.IHadoopClassFactory;
 import edu.uci.ics.hyracks.dataflow.std.base.IOpenableDataWriterOperator;
-import edu.uci.ics.hyracks.dataflow.std.group.IGroupAggregator;
 import edu.uci.ics.hyracks.dataflow.std.group.DeserializedPreclusteredGroupOperator;
+import edu.uci.ics.hyracks.dataflow.std.group.IGroupAggregator;
 import edu.uci.ics.hyracks.dataflow.std.util.DeserializedOperatorNodePushable;
 
 public class HadoopReducerOperatorDescriptor<K2, V2, K3, V3> extends AbstractHadoopOperatorDescriptor {
@@ -179,26 +178,19 @@
 
     private static final long serialVersionUID = 1L;
     private Class<? extends Reducer> reducerClass;
-    private static final String reducerClassKey = "mapred.reducer.class";
-    private static final String comparatorClassKey = "mapred.output.value.groupfn.class";
     private IComparatorFactory comparatorFactory;
 
-    public HadoopReducerOperatorDescriptor(JobSpecification spec, IComparatorFactory comparatorFactory,
-            Class<? extends Reducer> reducerClass, RecordDescriptor recordDescriptor, JobConf jobConf) {
-        super(spec, recordDescriptor, jobConf, new ClasspathBasedHadoopClassFactory());
+    public HadoopReducerOperatorDescriptor(JobSpecification spec, JobConf conf, IComparatorFactory comparatorFactory,
+            IHadoopClassFactory classFactory) {
+        super(spec, getRecordDescriptor(conf, classFactory), conf, classFactory);
         this.comparatorFactory = comparatorFactory;
-        this.reducerClass = reducerClass;
-    }
-
-    public HadoopReducerOperatorDescriptor(JobSpecification spec, JobConf conf, IHadoopClassFactory classFactory) {
-        super(spec, null, conf, classFactory);
     }
 
     private Reducer<K2, V2, K3, V3> createReducer() throws Exception {
         if (reducerClass != null) {
             return reducerClass.newInstance();
         } else {
-            Object reducer = getHadoopClassFactory().createReducer(getJobConfMap().get(reducerClassKey));
+            Object reducer = getHadoopClassFactory().createReducer(getJobConf().getReducerClass().getName());
             reducerClass = (Class<? extends Reducer>) reducer.getClass();
             return (Reducer) reducer;
         }
@@ -209,14 +201,14 @@
             IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
         try {
             if (this.comparatorFactory == null) {
-                String comparatorClassName = getJobConfMap().get(comparatorClassKey);
+                String comparatorClassName = getJobConf().getOutputValueGroupingComparator().getClass().getName();
                 RawComparator rawComparator = null;
                 if (comparatorClassName != null) {
                     Class comparatorClazz = getHadoopClassFactory().loadClass(comparatorClassName);
                     this.comparatorFactory = new KeyComparatorFactory(comparatorClazz);
 
                 } else {
-                    String mapOutputKeyClass = getJobConfMap().get("mapred.mapoutput.key.class");
+                    String mapOutputKeyClass = getJobConf().getMapOutputKeyClass().getName();
                     if (getHadoopClassFactory() != null) {
                         rawComparator = WritableComparator.get(getHadoopClassFactory().loadClass(mapOutputKeyClass));
                     } else {
@@ -235,28 +227,18 @@
         }
     }
 
-    public Class<? extends Reducer> getReducerClass() {
-        return reducerClass;
-    }
-
-    public void setReducerClass(Class<? extends Reducer> reducerClass) {
-        this.reducerClass = reducerClass;
-    }
-
-    @Override
-    public RecordDescriptor getRecordDescriptor(JobConf conf) {
-        String outputKeyClassName = conf.get("mapred.output.key.class");
-        String outputValueClassName = conf.get("mapred.output.value.class");
+    public static RecordDescriptor getRecordDescriptor(JobConf conf, IHadoopClassFactory classFactory) {
+        String outputKeyClassName = conf.getOutputKeyClass().getName();
+        String outputValueClassName = conf.getOutputValueClass().getName();
         RecordDescriptor recordDescriptor = null;
         try {
-            if (getHadoopClassFactory() == null) {
-                recordDescriptor = DatatypeHelper.createKeyValueRecordDescriptor(
-                        (Class<? extends Writable>) Class.forName(outputKeyClassName),
-                        (Class<? extends Writable>) Class.forName(outputValueClassName));
+            if (classFactory == null) {
+                recordDescriptor = DatatypeHelper.createKeyValueRecordDescriptor((Class<? extends Writable>) Class
+                        .forName(outputKeyClassName), (Class<? extends Writable>) Class.forName(outputValueClassName));
             } else {
                 recordDescriptor = DatatypeHelper.createKeyValueRecordDescriptor(
-                        (Class<? extends Writable>) getHadoopClassFactory().loadClass(outputKeyClassName),
-                        (Class<? extends Writable>) getHadoopClassFactory().loadClass(outputValueClassName));
+                        (Class<? extends Writable>) classFactory.loadClass(outputKeyClassName),
+                        (Class<? extends Writable>) classFactory.loadClass(outputValueClassName));
             }
         } catch (Exception e) {
             e.printStackTrace();
diff --git a/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopWriteOperatorDescriptor.java b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopWriteOperatorDescriptor.java
new file mode 100644
index 0000000..090bca1
--- /dev/null
+++ b/hyracks-dataflow-hadoop/src/main/java/edu/uci/ics/hyracks/dataflow/hadoop/HadoopWriteOperatorDescriptor.java
@@ -0,0 +1,171 @@
+/*
+ * Copyright 2009-2010 by The Regents of the University of California
+ * Licensed 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 from
+ * 
+ *     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 edu.uci.ics.hyracks.dataflow.hadoop;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.Writer;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.GzipCodec;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+
+import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.hadoop.util.DatatypeHelper;
+import edu.uci.ics.hyracks.dataflow.std.file.AbstractFileWriteOperatorDescriptor;
+import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
+import edu.uci.ics.hyracks.dataflow.std.file.IRecordWriter;
+import edu.uci.ics.hyracks.dataflow.std.file.RecordWriter;
+
+public class HadoopWriteOperatorDescriptor extends AbstractFileWriteOperatorDescriptor {
+
+    private static String nullWritableClassName = NullWritable.class.getName();
+
+    private static class HDFSWriter extends RecordWriter {
+
+        HDFSWriter(FileSystem fileSystem, String hdfsPath, int[] columns, char separator) throws Exception {
+            super(columns, separator, new Object[] { fileSystem, hdfsPath });
+        }
+
+        @Override
+        public OutputStream createOutputStream(Object[] args) throws Exception {
+            FSDataOutputStream fs = ((FileSystem) args[0]).create(new Path((String) args[1]));
+            return fs;
+        }
+
+        @Override
+        public void write(Object[] record) throws Exception {
+            if (!nullWritableClassName.equals((record[0].getClass().getName()))) {
+                bufferedWriter.write(String.valueOf(record[0]));
+            }
+            if (!nullWritableClassName.equals((record[1].getClass().getName()))) {
+                bufferedWriter.write(separator);
+                bufferedWriter.write(String.valueOf(record[1]));
+            }
+            bufferedWriter.write("\n");
+        }
+    }
+
+    private static class HDFSSequenceWriter extends RecordWriter {
+        private Writer writer;
+
+        HDFSSequenceWriter(FileSystem fileSystem, String hdfsPath, Writer writer) throws Exception {
+            super(null, COMMA, new Object[] { fileSystem, hdfsPath });
+            this.writer = writer;
+        }
+
+        @Override
+        public OutputStream createOutputStream(Object[] args) throws Exception {
+            return null;
+        }
+
+        @Override
+        public void close() {
+            try {
+                writer.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+
+        @Override
+        public void write(Object[] record) throws Exception {
+            Object key = record[0];
+            Object value = record[1];
+            writer.append(key, value);
+        }
+    }
+
+    private static final long serialVersionUID = 1L;
+    private static final char COMMA = ',';
+    private boolean sequenceFileOutput = false;
+    Map<String, String> jobConfMap;
+
+    @Override
+    protected IRecordWriter createRecordWriter(File file, int index) throws Exception {
+        JobConf conf = DatatypeHelper.map2JobConf((HashMap) jobConfMap);
+        conf.setClassLoader(this.getClass().getClassLoader());
+        FileSystem fileSystem = null;
+        try {
+            fileSystem = FileSystem.get(conf);
+        } catch (IOException ioe) {
+            ioe.printStackTrace();
+        }
+        Path path = new Path(file.getPath());
+        checkIfCanWriteToHDFS(new FileSplit[] { new FileSplit("localhost", file) });
+        FSDataOutputStream outputStream = fileSystem.create(path);
+        outputStream.close();
+        if (sequenceFileOutput) {
+            Class keyClass = Class.forName(conf.getOutputKeyClass().getName());
+            Class valueClass = Class.forName(conf.getOutputValueClass().getName());
+            conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);
+            Writer writer = SequenceFile.createWriter(fileSystem, conf, path, keyClass, valueClass);
+            return new HDFSSequenceWriter(fileSystem, file.getAbsolutePath(), writer);
+        } else {
+            return new HDFSWriter(fileSystem, file.getPath(), null, COMMA);
+        }
+    }
+
+    private boolean checkIfCanWriteToHDFS(FileSplit[] fileSplits) throws Exception {
+        boolean canWrite = true;
+        JobConf conf = DatatypeHelper.map2JobConf((HashMap) jobConfMap);
+        FileSystem fileSystem = null;
+        try {
+            fileSystem = FileSystem.get(conf);
+            for (FileSplit fileSplit : fileSplits) {
+                Path path = new Path(fileSplit.getLocalFile().getPath());
+                canWrite = !fileSystem.exists(path);
+                if (!canWrite) {
+                    throw new Exception(" Output path :  already exists : " + path);
+                }
+            }
+        } catch (IOException ioe) {
+            ioe.printStackTrace();
+            throw ioe;
+        }
+        return canWrite;
+    }
+
+    private static FileSplit[] getOutputSplits(JobConf conf, int noOfMappers) {
+        int numOutputters = conf.getNumReduceTasks() != 0 ? conf.getNumReduceTasks() : noOfMappers;
+        FileSplit[] outputFileSplits = new FileSplit[numOutputters];
+        String absolutePath = FileOutputFormat.getOutputPath(conf).toString();
+        for (int index = 0; index < numOutputters; index++) {
+            String suffix = new String("part-00000");
+            suffix = new String(suffix.substring(0, suffix.length() - ("" + index).length()));
+            suffix = suffix + index;
+            String outputPath = absolutePath + "/" + suffix;
+            outputFileSplits[index] = new FileSplit("localhost", new File(outputPath));
+        }
+        return outputFileSplits;
+    }
+
+    public HadoopWriteOperatorDescriptor(JobSpecification jobSpec, JobConf jobConf, int numMapTasks) throws Exception {
+        super(jobSpec, getOutputSplits(jobConf, numMapTasks));
+        this.jobConfMap = DatatypeHelper.jobConf2Map(jobConf);
+        checkIfCanWriteToHDFS(super.splits);
+        this.sequenceFileOutput = (jobConf.getOutputFormat() instanceof SequenceFileOutputFormat);
+    }
+}
diff --git a/hyracks-dataflow-hadoop/target/maven-archiver/pom.properties b/hyracks-dataflow-hadoop/target/maven-archiver/pom.properties
new file mode 100644
index 0000000..1c664bc
--- /dev/null
+++ b/hyracks-dataflow-hadoop/target/maven-archiver/pom.properties
@@ -0,0 +1,5 @@
+#Generated by Maven
+#Tue Oct 12 17:32:14 PDT 2010
+version=0.1.3-SNAPSHOT
+groupId=edu.uci.ics.hyracks
+artifactId=hyracks-dataflow-hadoop