Extensive changes to the underlying code base.

Fully tested and working support for region level spliting
Reduced number of mappers.

1 files changed, 22 insertions, 3 deletions

diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseTap.java b/src/main/java/parallelai/spyglass/hbase/HBaseTap.java
index 07b5aa7..bfe6670 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseTap.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseTap.java
@@ -12,6 +12,8 @@
 
 package parallelai.spyglass.hbase;
 
+import parallelai.spyglass.hbase.HBaseConstants.SplitType;
+
 import parallelai.spyglass.hbase.HBaseConstants.SourceMode;
 import cascading.flow.FlowProcess;
 import cascading.tap.SinkMode;
@@ -63,6 +65,8 @@ public class HBaseTap extends Tap<JobConf, RecordReader, OutputCollector> {
   /** Field tableName */
   private String tableName;
 
+  private SplitType splitType = SplitType.GRANULAR;
+
   /**
    * Constructor HBaseTap creates a new HBaseTap instance.
    *
@@ -204,7 +208,7 @@ public class HBaseTap extends Tap<JobConf, RecordReader, OutputCollector> {
       return true;
     }
 
-    LOG.info("creating hbase table: {}", tableName);
+    LOG.info("Creating HBase Table: {}", tableName);
 
     HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
 
@@ -256,8 +260,19 @@ public class HBaseTap extends Tap<JobConf, RecordReader, OutputCollector> {
 //    process.getID();
 //    
 //    super.getFullIdentifier(conf);
-    
-    HBaseInputFormat.setTableName(conf, tableName);
+
+    switch(splitType) {
+        case GRANULAR:
+            HBaseInputFormatGranular.setTableName(conf, tableName);
+            break;
+
+        case REGIONAL:
+            HBaseInputFormatRegional.setTableName(conf, tableName);
+            break;
+
+        default:
+            LOG.error("Unknown Split Type : " + splitType);
+    }
     
     for( SourceConfig sc : sourceConfigList) {
       sc.configure(conf);
@@ -266,6 +281,10 @@ public class HBaseTap extends Tap<JobConf, RecordReader, OutputCollector> {
     super.sourceConfInit(process, conf);
   }
 
+  public void setInputSplitType(SplitType sType) {
+      this.splitType = sType;
+  }
+
   @Override
   public boolean equals(Object object) {
     if (this == object) {