sqoop-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From b...@apache.org
Subject [4/5] SQOOP-666 Introduce execution engine (Jarek Jarcec Cecho)
Date Tue, 06 Nov 2012 20:53:35 GMT
http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopOutputFormatLoadExecutor.java
----------------------------------------------------------------------
diff --git a/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopOutputFormatLoadExecutor.java b/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopOutputFormatLoadExecutor.java
new file mode 100644
index 0000000..96e1533
--- /dev/null
+++ b/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopOutputFormatLoadExecutor.java
@@ -0,0 +1,228 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *     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 org.apache.sqoop.job.mr;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.sqoop.common.SqoopException;
+import org.apache.sqoop.core.CoreError;
+import org.apache.sqoop.job.JobConstants;
+import org.apache.sqoop.job.PrefixContext;
+import org.apache.sqoop.job.etl.Loader;
+import org.apache.sqoop.job.io.Data;
+import org.apache.sqoop.job.io.DataReader;
+import org.apache.sqoop.utils.ClassUtils;
+
+public class SqoopOutputFormatLoadExecutor {
+
+  public static final Log LOG =
+      LogFactory.getLog(SqoopOutputFormatLoadExecutor.class.getName());
+
+  private boolean readerFinished;
+  private boolean writerFinished;
+  private Data data;
+  private JobContext context;
+  private SqoopRecordWriter producer;
+  private ConsumerThread consumer;
+
+  public SqoopOutputFormatLoadExecutor(JobContext jobctx) {
+    data = new Data();
+    context = jobctx;
+    producer = new SqoopRecordWriter();
+    consumer = new ConsumerThread();
+  }
+
+  public RecordWriter<Data, NullWritable> getRecordWriter() {
+    consumer.setDaemon(true);
+    consumer.start();
+    return producer;
+  }
+
+  public class SqoopRecordWriter extends RecordWriter<Data, NullWritable> {
+    @Override
+    public void write(Data key, NullWritable value) {
+      synchronized (data) {
+        if (readerFinished) {
+          consumer.checkException();
+          return;
+        }
+
+        try {
+          if (!data.isEmpty()) {
+            // wait for reader to consume data
+            data.wait();
+          }
+
+          int type = key.getType();
+          data.setContent(key.getContent(type), type);
+
+          // notify reader that the data is ready
+          data.notify();
+
+        } catch (InterruptedException e) {
+          // inform reader that writer is finished
+          writerFinished = true;
+
+          // unlock reader so it can continue
+          data.notify();
+
+          // throw exception
+          throw new SqoopException(CoreError.CORE_0015, e);
+        }
+      }
+    }
+
+    @Override
+    public void close(TaskAttemptContext context) {
+      synchronized (data) {
+        if (readerFinished) {
+          consumer.checkException();
+          return;
+        }
+
+        try {
+          if (!data.isEmpty()) {
+            // wait for reader to consume data
+            data.wait();
+          }
+
+          writerFinished = true;
+
+          data.notify();
+
+        } catch (InterruptedException e) {
+          // inform reader that writer is finished
+          writerFinished = true;
+
+          // unlock reader so it can continue
+          data.notify();
+
+          // throw exception
+          throw new SqoopException(CoreError.CORE_0015, e);
+        }
+      }
+    }
+  }
+
+  public class OutputFormatDataReader extends DataReader {
+    @Override
+    public void setFieldDelimiter(char fieldDelimiter) {
+      data.setFieldDelimiter(fieldDelimiter);
+    }
+
+    @Override
+    public Object[] readArrayRecord() {
+      return (Object[])readContent(Data.ARRAY_RECORD);
+    }
+
+    @Override
+    public String readCsvRecord() {
+      return (String)readContent(Data.CSV_RECORD);
+    }
+
+    @Override
+    public Object readContent(int type) {
+      synchronized (data) {
+        if (writerFinished) {
+          return null;
+        }
+
+        try {
+          if (data.isEmpty()) {
+            // wait for writer to produce data
+            data.wait();
+          }
+
+          Object content = data.getContent(type);
+          data.setContent(null, Data.EMPTY_DATA);
+
+          // notify writer that data is consumed
+          data.notify();
+
+          return content;
+
+        } catch (InterruptedException e) {
+          // inform writer that reader is finished
+          readerFinished = true;
+
+          // unlock writer so it can continue
+          data.notify();
+
+          // throw exception
+          throw new SqoopException(CoreError.CORE_0016, e);
+        }
+      }
+    }
+  }
+
+  public class ConsumerThread extends Thread {
+    private SqoopException exception = null;
+
+    public void checkException() {
+      if (exception != null) {
+        throw exception;
+      }
+    }
+
+    @Override
+    public void run() {
+      DataReader reader = new OutputFormatDataReader();
+
+      Configuration conf = context.getConfiguration();
+
+
+      String loaderName = conf.get(JobConstants.JOB_ETL_LOADER);
+      Loader loader = (Loader) ClassUtils.instantiate(loaderName);
+
+      // Get together framework context as configuration prefix by nothing
+      PrefixContext frameworkContext = new PrefixContext(conf, "");
+
+      try {
+        loader.run(frameworkContext, reader);
+      } catch (Throwable t) {
+        throw new SqoopException(CoreError.CORE_0018, t);
+      }
+
+      synchronized (data) {
+        // inform writer that reader is finished
+        readerFinished = true;
+
+        // unlock writer so it can continue
+        data.notify();
+
+        // if no exception happens yet
+        if (exception == null && !writerFinished) {
+          // create exception if data are not all consumed
+          exception = new SqoopException(CoreError.CORE_0019);
+        }
+
+        // throw deferred exception if exist
+        if (exception != null) {
+          throw exception;
+        }
+      }
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopReducer.java
----------------------------------------------------------------------
diff --git a/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopReducer.java b/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopReducer.java
new file mode 100644
index 0000000..d236148
--- /dev/null
+++ b/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopReducer.java
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *     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 org.apache.sqoop.job.mr;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.sqoop.job.io.Data;
+
+/**
+ * A reducer to perform reduce function.
+ */
+public class SqoopReducer
+    extends Reducer<Data, NullWritable, Data, NullWritable> {
+
+  public static final Log LOG =
+      LogFactory.getLog(SqoopReducer.class.getName());
+
+}

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopSplit.java
----------------------------------------------------------------------
diff --git a/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopSplit.java b/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopSplit.java
new file mode 100644
index 0000000..7dc9541
--- /dev/null
+++ b/execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopSplit.java
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *     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 org.apache.sqoop.job.mr;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.sqoop.common.SqoopException;
+import org.apache.sqoop.core.CoreError;
+import org.apache.sqoop.job.etl.Partition;
+import org.apache.sqoop.utils.ClassUtils;
+
+/**
+ * An input split to be read.
+ */
+public class SqoopSplit extends InputSplit implements Writable {
+
+  private Partition partition;
+
+  public void setPartition(Partition partition) {
+    this.partition = partition;
+  }
+
+  public Partition getPartition() {
+    return partition;
+  }
+
+  @Override
+  public long getLength() throws IOException {
+    return 0;
+  }
+
+  @Override
+  public String[] getLocations() throws IOException {
+    return new String[] {};
+  }
+
+  @Override
+  public void readFields(DataInput in) throws IOException {
+    // read Partition class name
+    String className = in.readUTF();
+    // instantiate Partition object
+    Class<?> clz = ClassUtils.loadClass(className);
+    if (clz == null) {
+      throw new SqoopException(CoreError.CORE_0009, className);
+    }
+    try {
+      partition = (Partition) clz.newInstance();
+    } catch (Exception e) {
+      throw new SqoopException(CoreError.CORE_0010, className, e);
+    }
+    // read Partition object content
+    partition.readFields(in);
+  }
+
+  @Override
+  public void write(DataOutput out) throws IOException {
+    // write Partition class name
+    out.writeUTF(partition.getClass().getName());
+    // write Partition object content
+    partition.write(out);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/mapreduce/src/test/java/org/apache/sqoop/job/FileUtils.java
----------------------------------------------------------------------
diff --git a/execution/mapreduce/src/test/java/org/apache/sqoop/job/FileUtils.java b/execution/mapreduce/src/test/java/org/apache/sqoop/job/FileUtils.java
new file mode 100644
index 0000000..e685883
--- /dev/null
+++ b/execution/mapreduce/src/test/java/org/apache/sqoop/job/FileUtils.java
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *     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 org.apache.sqoop.job;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+public class FileUtils {
+
+  public static boolean exists(String file) throws IOException {
+    Path path = new Path(file);
+    FileSystem fs = path.getFileSystem(new Configuration());
+    return fs.exists(path);
+  }
+
+  public static void delete(String file) throws IOException {
+    Path path = new Path(file);
+    FileSystem fs = path.getFileSystem(new Configuration());
+    if (fs.exists(path)) {
+      fs.delete(path, true);
+    }
+  }
+
+  public static void mkdirs(String directory) throws IOException {
+    Path path = new Path(directory);
+    FileSystem fs = path.getFileSystem(new Configuration());
+    if (!fs.exists(path)) {
+      fs.mkdirs(path);
+    }
+  }
+
+  public static InputStream open(String fileName)
+    throws IOException, ClassNotFoundException {
+    Path filepath = new Path(fileName);
+    FileSystem fs = filepath.getFileSystem(new Configuration());
+    return fs.open(filepath);
+  }
+
+  public static OutputStream create(String fileName) throws IOException {
+    Path filepath = new Path(fileName);
+    FileSystem fs = filepath.getFileSystem(new Configuration());
+    return fs.create(filepath, false);
+  }
+
+  private FileUtils() {
+    // Disable explicit object creation
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/mapreduce/src/test/java/org/apache/sqoop/job/JobUtils.java
----------------------------------------------------------------------
diff --git a/execution/mapreduce/src/test/java/org/apache/sqoop/job/JobUtils.java b/execution/mapreduce/src/test/java/org/apache/sqoop/job/JobUtils.java
new file mode 100644
index 0000000..e6ead3f
--- /dev/null
+++ b/execution/mapreduce/src/test/java/org/apache/sqoop/job/JobUtils.java
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *     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 org.apache.sqoop.job;
+
+import java.io.IOException;
+
+import junit.framework.Assert;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.sqoop.job.io.Data;
+import org.apache.sqoop.job.mr.SqoopFileOutputFormat;
+import org.apache.sqoop.job.mr.SqoopInputFormat;
+import org.apache.sqoop.job.mr.SqoopMapper;
+import org.apache.sqoop.job.mr.SqoopNullOutputFormat;
+import org.apache.sqoop.job.mr.SqoopSplit;
+
+public class JobUtils {
+
+  public static void runJob(Configuration conf)
+      throws IOException, InterruptedException, ClassNotFoundException {
+    runJob(conf, SqoopInputFormat.class, SqoopMapper.class,
+        (conf.get(FileOutputFormat.OUTDIR) != null) ?
+        SqoopFileOutputFormat.class : SqoopNullOutputFormat.class);
+  }
+
+  public static void runJob(Configuration conf,
+      Class<? extends InputFormat<SqoopSplit, NullWritable>> input,
+      Class<? extends Mapper<SqoopSplit, NullWritable, Data, NullWritable>> mapper,
+      Class<? extends OutputFormat<Data, NullWritable>> output)
+      throws IOException, InterruptedException, ClassNotFoundException {
+    Job job = Job.getInstance(conf);
+    job.setInputFormatClass(input);
+    job.setMapperClass(mapper);
+    job.setMapOutputKeyClass(Data.class);
+    job.setMapOutputValueClass(NullWritable.class);
+    job.setOutputFormatClass(output);
+    job.setOutputKeyClass(Data.class);
+    job.setOutputValueClass(NullWritable.class);
+
+    boolean success = job.waitForCompletion(true);
+    Assert.assertEquals("Job failed!", true, success);
+  }
+
+  private JobUtils() {
+    // Disable explicit object creation
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestHdfsLoad.java
----------------------------------------------------------------------
diff --git a/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestHdfsLoad.java b/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestHdfsLoad.java
new file mode 100644
index 0000000..c74faa2
--- /dev/null
+++ b/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestHdfsLoad.java
@@ -0,0 +1,232 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *     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 org.apache.sqoop.job;
+
+import java.io.BufferedReader;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.LinkedList;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.sqoop.job.etl.Extractor;
+import org.apache.sqoop.job.etl.HdfsSequenceImportLoader;
+import org.apache.sqoop.job.etl.HdfsTextImportLoader;
+import org.apache.sqoop.job.etl.Partition;
+import org.apache.sqoop.job.etl.Partitioner;
+import org.apache.sqoop.job.io.Data;
+import org.apache.sqoop.job.io.DataWriter;
+import org.apache.sqoop.job.mr.SqoopFileOutputFormat;
+import org.junit.Test;
+
+public class TestHdfsLoad extends TestCase {
+
+  private static final String OUTPUT_ROOT = "/tmp/sqoop/warehouse/";
+  private static final String OUTPUT_FILE = "part-r-00000";
+  private static final int START_ID = 1;
+  private static final int NUMBER_OF_IDS = 9;
+  private static final int NUMBER_OF_ROWS_PER_ID = 10;
+
+  private String outdir;
+
+  public TestHdfsLoad() {
+    outdir = OUTPUT_ROOT + "/" + getClass().getSimpleName();
+  }
+
+  public void testVoid() {}
+  /*
+  @Test
+  public void testUncompressedText() throws Exception {
+    FileUtils.delete(outdir);
+
+    Configuration conf = new Configuration();
+    conf.set(JobConstants.JOB_ETL_PARTITIONER, DummyPartitioner.class.getName());
+    conf.set(JobConstants.JOB_ETL_EXTRACTOR, DummyExtractor.class.getName());
+    conf.set(JobConstants.JOB_ETL_LOADER, HdfsTextImportLoader.class.getName());
+    conf.set(FileOutputFormat.OUTDIR, outdir);
+    JobUtils.runJob(conf);
+
+    String fileName = outdir + "/" +  OUTPUT_FILE;
+    InputStream filestream = FileUtils.open(fileName);
+    BufferedReader filereader = new BufferedReader(new InputStreamReader(
+        filestream, Data.CHARSET_NAME));
+    verifyOutputText(filereader);
+  }
+
+  @Test
+  public void testCompressedText() throws Exception {
+    FileUtils.delete(outdir);
+
+    Configuration conf = new Configuration();
+    conf.set(JobConstants.JOB_ETL_PARTITIONER, DummyPartitioner.class.getName());
+    conf.set(JobConstants.JOB_ETL_EXTRACTOR, DummyExtractor.class.getName());
+    conf.set(JobConstants.JOB_ETL_LOADER, HdfsTextImportLoader.class.getName());
+    conf.set(FileOutputFormat.OUTDIR, outdir);
+    conf.setBoolean(FileOutputFormat.COMPRESS, true);
+    JobUtils.runJob(conf);
+
+    Class<? extends CompressionCodec> codecClass = conf.getClass(
+        FileOutputFormat.COMPRESS_CODEC, SqoopFileOutputFormat.DEFAULT_CODEC)
+        .asSubclass(CompressionCodec.class);
+    CompressionCodec codec = ReflectionUtils.newInstance(codecClass, conf);
+    String fileName = outdir + "/" +  OUTPUT_FILE + codec.getDefaultExtension();
+    InputStream filestream = codec.createInputStream(FileUtils.open(fileName));
+    BufferedReader filereader = new BufferedReader(new InputStreamReader(
+        filestream, Data.CHARSET_NAME));
+    verifyOutputText(filereader);
+  }
+
+  private void verifyOutputText(BufferedReader reader) throws IOException {
+    String actual = null;
+    String expected;
+    Data data = new Data();
+    int index = START_ID*NUMBER_OF_ROWS_PER_ID;
+    while ((actual = reader.readLine()) != null){
+      data.setContent(new Object[] {
+          new Integer(index), new Double(index), String.valueOf(index) },
+          Data.ARRAY_RECORD);
+      expected = data.toString();
+      index++;
+
+      assertEquals(expected, actual);
+    }
+    reader.close();
+
+    assertEquals(NUMBER_OF_IDS*NUMBER_OF_ROWS_PER_ID,
+        index-START_ID*NUMBER_OF_ROWS_PER_ID);
+  }
+
+  @Test
+  public void testUncompressedSequence() throws Exception {
+    FileUtils.delete(outdir);
+
+    Configuration conf = new Configuration();
+    conf.set(JobConstants.JOB_ETL_PARTITIONER, DummyPartitioner.class.getName());
+    conf.set(JobConstants.JOB_ETL_EXTRACTOR, DummyExtractor.class.getName());
+    conf.set(JobConstants.JOB_ETL_LOADER, HdfsSequenceImportLoader.class.getName());
+    conf.set(FileOutputFormat.OUTDIR, outdir);
+    JobUtils.runJob(conf);
+
+    Path filepath = new Path(outdir,
+        OUTPUT_FILE + HdfsSequenceImportLoader.EXTENSION);
+    SequenceFile.Reader filereader = new SequenceFile.Reader(conf,
+        SequenceFile.Reader.file(filepath));
+    verifyOutputSequence(filereader);
+  }
+
+  @Test
+  public void testCompressedSequence() throws Exception {
+    FileUtils.delete(outdir);
+
+    Configuration conf = new Configuration();
+    conf.set(JobConstants.JOB_ETL_PARTITIONER, DummyPartitioner.class.getName());
+    conf.set(JobConstants.JOB_ETL_EXTRACTOR, DummyExtractor.class.getName());
+    conf.set(JobConstants.JOB_ETL_LOADER, HdfsSequenceImportLoader.class.getName());
+    conf.set(FileOutputFormat.OUTDIR, outdir);
+    conf.setBoolean(FileOutputFormat.COMPRESS, true);
+    JobUtils.runJob(conf);
+
+    Path filepath = new Path(outdir,
+        OUTPUT_FILE + HdfsSequenceImportLoader.EXTENSION);
+    SequenceFile.Reader filereader = new SequenceFile.Reader(conf,
+        SequenceFile.Reader.file(filepath));
+    verifyOutputSequence(filereader);
+  }
+
+  private void verifyOutputSequence(SequenceFile.Reader reader) throws IOException {
+    int index = START_ID*NUMBER_OF_ROWS_PER_ID;
+    Text actual = new Text();
+    Text expected = new Text();
+    Data data = new Data();
+    while (reader.next(actual)){
+      data.setContent(new Object[] {
+          new Integer(index), new Double(index), String.valueOf(index) },
+          Data.ARRAY_RECORD);
+      expected.set(data.toString());
+      index++;
+
+      assertEquals(expected.toString(), actual.toString());
+    }
+    reader.close();
+
+    assertEquals(NUMBER_OF_IDS*NUMBER_OF_ROWS_PER_ID,
+        index-START_ID*NUMBER_OF_ROWS_PER_ID);
+  }
+
+  public static class DummyPartition extends Partition {
+    private int id;
+
+    public void setId(int id) {
+      this.id = id;
+    }
+
+    public int getId() {
+      return id;
+    }
+
+    @Override
+    public void readFields(DataInput in) throws IOException {
+      id = in.readInt();
+    }
+
+    @Override
+    public void write(DataOutput out) throws IOException {
+      out.writeInt(id);
+    }
+  }
+
+  public static class DummyPartitioner extends Partitioner {
+    @Override
+    public List<Partition> initialize(Context context) {
+      List<Partition> partitions = new LinkedList<Partition>();
+      for (int id = START_ID; id <= NUMBER_OF_IDS; id++) {
+        DummyPartition partition = new DummyPartition();
+        partition.setId(id);
+        partitions.add(partition);
+      }
+      return partitions;
+    }
+  }
+
+  public static class DummyExtractor extends Extractor {
+    @Override
+    public void initialize(Context context, Partition partition, DataWriter writer) {
+      int id = ((DummyPartition)partition).getId();
+      for (int row = 0; row < NUMBER_OF_ROWS_PER_ID; row++) {
+        Object[] array = new Object[] {
+          new Integer(id*NUMBER_OF_ROWS_PER_ID+row),
+          new Double(id*NUMBER_OF_ROWS_PER_ID+row),
+          String.valueOf(id*NUMBER_OF_ROWS_PER_ID+row)
+        };
+        writer.writeArrayRecord(array);
+      }
+    }
+  }
+  */
+}

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestJobEngine.java
----------------------------------------------------------------------
diff --git a/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestJobEngine.java b/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestJobEngine.java
new file mode 100644
index 0000000..51dddb4
--- /dev/null
+++ b/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestJobEngine.java
@@ -0,0 +1,196 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *     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 org.apache.sqoop.job;
+
+import java.io.BufferedReader;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Locale;
+import java.util.ResourceBundle;
+
+import junit.framework.TestCase;
+
+import org.apache.sqoop.connector.spi.SqoopConnector;
+import org.apache.sqoop.job.etl.Exporter;
+import org.apache.sqoop.job.etl.Extractor;
+import org.apache.sqoop.job.etl.Importer;
+import org.apache.sqoop.job.etl.Initializer;
+import org.apache.sqoop.job.etl.Partition;
+import org.apache.sqoop.job.etl.Partitioner;
+import org.apache.sqoop.job.io.Data;
+import org.apache.sqoop.job.io.DataWriter;
+import org.apache.sqoop.model.MConnectionForms;
+import org.apache.sqoop.model.MJob.Type;
+import org.apache.sqoop.model.MJobForms;
+import org.apache.sqoop.validation.Validator;
+import org.junit.Test;
+
+public class TestJobEngine extends TestCase {
+
+  private static final String DATA_DIR = TestJobEngine.class.getSimpleName();
+  private static final String WAREHOUSE_ROOT = "/tmp/sqoop/warehouse/";
+
+  private static final String OUTPUT_DIR = WAREHOUSE_ROOT + DATA_DIR;
+  private static final String OUTPUT_FILE = "part-r-00000";
+  private static final int START_PARTITION = 1;
+  private static final int NUMBER_OF_PARTITIONS = 9;
+  private static final int NUMBER_OF_ROWS_PER_PARTITION = 10;
+
+  public void testVoid() { }
+/*
+  @Test
+  public void testImport() throws Exception {
+    FileUtils.delete(OUTPUT_DIR);
+
+    DummyConnector connector = new DummyConnector();
+    EtlOptions options = new EtlOptions(connector);
+
+    JobEngine engine = new JobEngine();
+    engine.initialize(options);
+
+    String fileName = OUTPUT_DIR + "/" + OUTPUT_FILE;
+    InputStream filestream = FileUtils.open(fileName);
+    BufferedReader filereader = new BufferedReader(new InputStreamReader(
+        filestream, Data.CHARSET_NAME));
+    verifyOutput(filereader);
+  }
+
+  private void verifyOutput(BufferedReader reader)
+      throws IOException {
+    String line = null;
+    int index = START_PARTITION*NUMBER_OF_ROWS_PER_PARTITION;
+    Data expected = new Data();
+    while ((line = reader.readLine()) != null){
+      expected.setContent(new Object[] {
+          new Integer(index),
+          new Double(index),
+          String.valueOf(index) },
+          Data.ARRAY_RECORD);
+      index++;
+
+      assertEquals(expected.toString(), line);
+    }
+    reader.close();
+
+    assertEquals(NUMBER_OF_PARTITIONS*NUMBER_OF_ROWS_PER_PARTITION,
+        index-START_PARTITION*NUMBER_OF_ROWS_PER_PARTITION);
+  }
+
+  public class DummyConnector implements SqoopConnector {
+
+    @Override
+    public Importer getImporter() {
+      return new Importer(
+          DummyImportInitializer.class,
+          DummyImportPartitioner.class,
+          DummyImportExtractor.class,
+          null);
+    }
+
+    @Override
+    public Exporter getExporter() {
+      fail("This method should not be invoked.");
+      return null;
+    }
+
+    @Override
+    public ResourceBundle getBundle(Locale locale) {
+      fail("This method should not be invoked.");
+      return null;
+    }
+
+    @Override
+    public Validator getValidator() {
+      fail("This method should not be invoked.");
+      return null;
+    }
+
+    @Override
+    public Class getConnectionConfigurationClass() {
+      fail("This method should not be invoked.");
+      return null;
+    }
+
+    @Override
+    public Class getJobConfigurationClass(Type jobType) {
+      fail("This method should not be invoked.");
+      return null;
+    }
+  }
+
+  public static class DummyImportInitializer extends Initializer {
+    @Override
+    public void initialize(MutableContext context, Options options) {
+      context.setString(Constants.JOB_ETL_OUTPUT_DIRECTORY, OUTPUT_DIR);
+    }
+  }
+
+  public static class DummyImportPartition extends Partition {
+    private int id;
+
+    public void setId(int id) {
+      this.id = id;
+    }
+
+    public int getId() {
+      return id;
+    }
+
+    @Override
+    public void readFields(DataInput in) throws IOException {
+      id = in.readInt();
+    }
+
+    @Override
+    public void write(DataOutput out) throws IOException {
+      out.writeInt(id);
+    }
+  }
+
+  public static class DummyImportPartitioner extends Partitioner {
+    @Override
+    public List<Partition> initialize(Context context) {
+      List<Partition> partitions = new LinkedList<Partition>();
+      for (int id = START_PARTITION; id <= NUMBER_OF_PARTITIONS; id++) {
+        DummyImportPartition partition = new DummyImportPartition();
+        partition.setId(id);
+        partitions.add(partition);
+      }
+      return partitions;
+    }
+  }
+
+  public static class DummyImportExtractor extends Extractor {
+    @Override
+    public void initialize(Context context, Partition partition, DataWriter writer) {
+      int id = ((DummyImportPartition)partition).getId();
+      for (int row = 0; row < NUMBER_OF_ROWS_PER_PARTITION; row++) {
+        writer.writeArrayRecord(new Object[] {
+            new Integer(id*NUMBER_OF_ROWS_PER_PARTITION+row),
+            new Double(id*NUMBER_OF_ROWS_PER_PARTITION+row),
+            String.valueOf(id*NUMBER_OF_ROWS_PER_PARTITION+row)});
+      }
+    }
+  }
+*/
+}

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestMapReduce.java
----------------------------------------------------------------------
diff --git a/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestMapReduce.java b/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestMapReduce.java
new file mode 100644
index 0000000..94ab560
--- /dev/null
+++ b/execution/mapreduce/src/test/java/org/apache/sqoop/job/TestMapReduce.java
@@ -0,0 +1,229 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *     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 org.apache.sqoop.job;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.LinkedList;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.sqoop.job.etl.Extractor;
+import org.apache.sqoop.job.etl.Loader;
+import org.apache.sqoop.job.etl.Partition;
+import org.apache.sqoop.job.etl.Partitioner;
+import org.apache.sqoop.job.io.Data;
+import org.apache.sqoop.job.io.DataReader;
+import org.apache.sqoop.job.io.DataWriter;
+import org.apache.sqoop.job.mr.SqoopInputFormat;
+import org.apache.sqoop.job.mr.SqoopMapper;
+import org.apache.sqoop.job.mr.SqoopNullOutputFormat;
+import org.apache.sqoop.job.mr.SqoopSplit;
+import org.junit.Test;
+
+public class TestMapReduce extends TestCase {
+
+  private static final int START_PARTITION = 1;
+  private static final int NUMBER_OF_PARTITIONS = 9;
+  private static final int NUMBER_OF_ROWS_PER_PARTITION = 10;
+
+  public void testVoid() {}
+
+  /*
+  @Test
+  public void testInputFormat() throws Exception {
+    Configuration conf = new Configuration();
+    conf.set(JobConstants.JOB_ETL_PARTITIONER, DummyPartitioner.class.getName());
+    Job job = Job.getInstance(conf);
+
+    SqoopInputFormat inputformat = new SqoopInputFormat();
+    List<InputSplit> splits = inputformat.getSplits(job);
+    assertEquals(9, splits.size());
+
+    for (int id = START_PARTITION; id <= NUMBER_OF_PARTITIONS; id++) {
+      SqoopSplit split = (SqoopSplit)splits.get(id-1);
+      DummyPartition partition = (DummyPartition)split.getPartition();
+      assertEquals(id, partition.getId());
+    }
+  }
+
+  @Test
+  public void testMapper() throws Exception {
+    Configuration conf = new Configuration();
+    conf.set(JobConstants.JOB_ETL_PARTITIONER, DummyPartitioner.class.getName());
+    conf.set(JobConstants.JOB_ETL_EXTRACTOR, DummyExtractor.class.getName());
+
+    JobUtils.runJob(conf, SqoopInputFormat.class, SqoopMapper.class,
+        DummyOutputFormat.class);
+  }
+
+  @Test
+  public void testOutputFormat() throws Exception {
+    Configuration conf = new Configuration();
+    conf.set(JobConstants.JOB_ETL_PARTITIONER, DummyPartitioner.class.getName());
+    conf.set(JobConstants.JOB_ETL_EXTRACTOR, DummyExtractor.class.getName());
+    conf.set(JobConstants.JOB_ETL_LOADER, DummyLoader.class.getName());
+
+    JobUtils.runJob(conf, SqoopInputFormat.class, SqoopMapper.class,
+        SqoopNullOutputFormat.class);
+  }
+
+  public static class DummyPartition extends Partition {
+    private int id;
+
+    public void setId(int id) {
+      this.id = id;
+    }
+
+    public int getId() {
+      return id;
+    }
+
+    @Override
+    public void readFields(DataInput in) throws IOException {
+      id = in.readInt();
+    }
+
+    @Override
+    public void write(DataOutput out) throws IOException {
+      out.writeInt(id);
+    }
+  }
+
+  public static class DummyPartitioner extends Partitioner {
+    @Override
+    public List<Partition> initialize(Context context) {
+      List<Partition> partitions = new LinkedList<Partition>();
+      for (int id = START_PARTITION; id <= NUMBER_OF_PARTITIONS; id++) {
+        DummyPartition partition = new DummyPartition();
+        partition.setId(id);
+        partitions.add(partition);
+      }
+      return partitions;
+    }
+  }
+
+  public static class DummyExtractor extends Extractor {
+    @Override
+    public void initialize(Context context, Partition partition, DataWriter writer) {
+      int id = ((DummyPartition)partition).getId();
+      for (int row = 0; row < NUMBER_OF_ROWS_PER_PARTITION; row++) {
+        writer.writeArrayRecord(new Object[] {
+            new Integer(id*NUMBER_OF_ROWS_PER_PARTITION+row),
+            new Double(id*NUMBER_OF_ROWS_PER_PARTITION+row),
+            String.valueOf(id*NUMBER_OF_ROWS_PER_PARTITION+row)});
+      }
+    }
+  }
+
+  public static class DummyOutputFormat
+      extends OutputFormat<Data, NullWritable> {
+    @Override
+    public void checkOutputSpecs(JobContext context) {
+      // do nothing
+    }
+
+    @Override
+    public RecordWriter<Data, NullWritable> getRecordWriter(
+        TaskAttemptContext context) {
+      return new DummyRecordWriter();
+    }
+
+    @Override
+    public OutputCommitter getOutputCommitter(TaskAttemptContext context) {
+      return new DummyOutputCommitter();
+    }
+
+    public static class DummyRecordWriter
+        extends RecordWriter<Data, NullWritable> {
+      private int index = START_PARTITION*NUMBER_OF_ROWS_PER_PARTITION;
+      private Data data = new Data();
+
+      @Override
+      public void write(Data key, NullWritable value) {
+        data.setContent(new Object[] {
+          new Integer(index),
+          new Double(index),
+          String.valueOf(index)},
+          Data.ARRAY_RECORD);
+        index++;
+
+        assertEquals(data.toString(), key.toString());
+      }
+
+      @Override
+      public void close(TaskAttemptContext context) {
+        // do nothing
+      }
+    }
+
+    public static class DummyOutputCommitter extends OutputCommitter {
+      @Override
+      public void setupJob(JobContext jobContext) { }
+
+      @Override
+      public void setupTask(TaskAttemptContext taskContext) { }
+
+      @Override
+      public void commitTask(TaskAttemptContext taskContext) { }
+
+      @Override
+      public void abortTask(TaskAttemptContext taskContext) { }
+
+      @Override
+      public boolean needsTaskCommit(TaskAttemptContext taskContext) {
+        return false;
+      }
+    }
+  }
+
+  public static class DummyLoader extends Loader {
+    private int index = START_PARTITION*NUMBER_OF_ROWS_PER_PARTITION;
+    private Data expected = new Data();
+    private Data actual = new Data();
+
+    @Override
+    public void initialize(Context context, DataReader reader) {
+      Object[] array;
+      while ((array = reader.readArrayRecord()) != null) {
+        actual.setContent(array, Data.ARRAY_RECORD);
+
+        expected.setContent(new Object[] {
+          new Integer(index),
+          new Double(index),
+          String.valueOf(index)},
+          Data.ARRAY_RECORD);
+        index++;
+
+        assertEquals(expected.toString(), actual.toString());
+      };
+    }
+  }
+  */
+}

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/mapreduce/src/test/java/org/apache/sqoop/job/io/TestData.java
----------------------------------------------------------------------
diff --git a/execution/mapreduce/src/test/java/org/apache/sqoop/job/io/TestData.java b/execution/mapreduce/src/test/java/org/apache/sqoop/job/io/TestData.java
new file mode 100644
index 0000000..d4a7d4d
--- /dev/null
+++ b/execution/mapreduce/src/test/java/org/apache/sqoop/job/io/TestData.java
@@ -0,0 +1,75 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 at
+ *
+ *     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 org.apache.sqoop.job.io;
+
+import java.util.Arrays;
+
+import junit.framework.TestCase;
+
+import org.junit.Test;
+
+public class TestData extends TestCase {
+
+  private static final double TEST_NUMBER = Math.PI + 100;
+  @Test
+  public void testArrayToCsv() throws Exception {
+    Data data = new Data();
+    String expected;
+    String actual;
+
+    // with special characters:
+    expected =
+        Long.valueOf((long)TEST_NUMBER) + "," +
+        Double.valueOf(TEST_NUMBER) + "," +
+        "'" + String.valueOf(TEST_NUMBER) + "\\',s'" + "," +
+        Arrays.toString(new byte[] {1, 2, 3, 4, 5});
+    data.setContent(new Object[] {
+        Long.valueOf((long)TEST_NUMBER),
+        Double.valueOf(TEST_NUMBER),
+        String.valueOf(TEST_NUMBER) + "',s",
+        new byte[] {1, 2, 3, 4, 5} },
+        Data.ARRAY_RECORD);
+    actual = (String)data.getContent(Data.CSV_RECORD);
+    assertEquals(expected, actual);
+
+    // with null characters:
+    expected =
+        Long.valueOf((long)TEST_NUMBER) + "," +
+        Double.valueOf(TEST_NUMBER) + "," +
+        "null" + "," +
+        Arrays.toString(new byte[] {1, 2, 3, 4, 5});
+    data.setContent(new Object[] {
+        Long.valueOf((long)TEST_NUMBER),
+        Double.valueOf(TEST_NUMBER),
+        null,
+        new byte[] {1, 2, 3, 4, 5} },
+        Data.ARRAY_RECORD);
+    actual = (String)data.getContent(Data.CSV_RECORD);
+    assertEquals(expected, actual);
+  }
+
+  public static void assertEquals(Object expected, Object actual) {
+    if (expected instanceof byte[]) {
+      assertEquals(Arrays.toString((byte[])expected),
+          Arrays.toString((byte[])actual));
+    } else {
+      TestCase.assertEquals(expected, actual);
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/execution/pom.xml
----------------------------------------------------------------------
diff --git a/execution/pom.xml b/execution/pom.xml
new file mode 100644
index 0000000..fb9f801
--- /dev/null
+++ b/execution/pom.xml
@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 at
+
+     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.
+--><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>org.apache</groupId>
+    <artifactId>sqoop</artifactId>
+    <version>2.0.0-SNAPSHOT</version>
+  </parent>
+
+  <groupId>org.apache.sqoop</groupId>
+  <artifactId>execution</artifactId>
+  <name>Sqoop Execution Engines</name>
+  <packaging>pom</packaging>
+
+  <modules>
+    <module>mapreduce</module>
+  </modules>
+
+</project>

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index a4915fd..4211333 100644
--- a/pom.xml
+++ b/pom.xml
@@ -220,6 +220,7 @@ limitations under the License.
     <module>client</module>
     <module>docs</module>
     <module>connector</module>
+    <module>execution</module>
     <module>submission</module>
     <module>dist</module>
   </modules>

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/submission/mapreduce/pom.xml
----------------------------------------------------------------------
diff --git a/submission/mapreduce/pom.xml b/submission/mapreduce/pom.xml
index 03c06c0..f8a7d3d 100644
--- a/submission/mapreduce/pom.xml
+++ b/submission/mapreduce/pom.xml
@@ -37,6 +37,12 @@ limitations under the License.
     </dependency>
 
     <dependency>
+      <groupId>org.apache.sqoop.execution</groupId>
+      <artifactId>sqoop-execution-mapreduce</artifactId>
+      <version>2.0.0-SNAPSHOT</version>
+    </dependency>
+
+    <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>

http://git-wip-us.apache.org/repos/asf/sqoop/blob/06e054bc/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java
----------------------------------------------------------------------
diff --git a/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java b/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java
index 94098de..b8415e3 100644
--- a/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java
+++ b/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java
@@ -28,6 +28,8 @@ import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 import org.apache.log4j.Logger;
 import org.apache.sqoop.common.MapContext;
 import org.apache.sqoop.common.SqoopException;
+import org.apache.sqoop.execution.mapreduce.MRSubmissionRequest;
+import org.apache.sqoop.execution.mapreduce.MapreduceExecutionEngine;
 import org.apache.sqoop.framework.SubmissionRequest;
 import org.apache.sqoop.framework.SubmissionEngine;
 import org.apache.sqoop.job.JobConstants;
@@ -116,8 +118,22 @@ public class MapreduceSubmissionEngine extends SubmissionEngine {
    * {@inheritDoc}
    */
   @Override
-  @SuppressWarnings("unchecked")
-  public boolean submit(SubmissionRequest request) {
+  public boolean isExecutionEngineSupported(Class executionEngineClass) {
+    if(executionEngineClass == MapreduceExecutionEngine.class) {
+      return true;
+    }
+
+    return false;
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public boolean submit(SubmissionRequest generalRequest) {
+    // We're supporting only map reduce jobs
+    MRSubmissionRequest request = (MRSubmissionRequest) generalRequest;
+
     // Clone global configuration
     Configuration configuration = new Configuration(globalConfiguration);
 


Mime
View raw message