From 3501e241a2313cf49c371630cb6ebe0c3a47e991 Mon Sep 17 00:00:00 2001
From: Chandan Rajah <crajah@parallelai.com>
Date: Wed, 4 Sep 2013 10:32:07 +0100
Subject: Extensive changes to the underlying code base. Fully tested and
 working support for region level spliting Reduced number of mappers.

---
 .../parallelai/spyglass/hbase/HBaseScheme.java     | 35 +++++++++++++++++++---
 1 file changed, 31 insertions(+), 4 deletions(-)

(limited to 'src/main/java/parallelai/spyglass/hbase/HBaseScheme.java')

diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseScheme.java b/src/main/java/parallelai/spyglass/hbase/HBaseScheme.java
index aa446c1..6f04f01 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseScheme.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseScheme.java
@@ -12,6 +12,7 @@
 
 package parallelai.spyglass.hbase;
 
+import parallelai.spyglass.hbase.HBaseConstants.SplitType;
 import cascading.flow.FlowProcess;
 import cascading.scheme.Scheme;
 import cascading.scheme.SinkCall;
@@ -65,6 +66,8 @@ public class HBaseScheme
 //  private transient byte[][] fields;
   
   private boolean useSalt = false;
+
+  private SplitType splitType = SplitType.GRANULAR;
  
 
   /**
@@ -279,11 +282,31 @@ public class HBaseScheme
   @Override
   public void sourceConfInit(FlowProcess<JobConf> process,
       Tap<JobConf, RecordReader, OutputCollector> tap, JobConf conf) {
-    conf.setInputFormat(HBaseInputFormat.class);
 
-    String columns = getColumns();
-    LOG.debug("sourcing from columns: {}", columns);
-    conf.set(HBaseInputFormat.COLUMN_LIST, columns);
+      switch(splitType) {
+          case GRANULAR:
+          {
+              conf.setInputFormat(HBaseInputFormatGranular.class);
+
+              String columns = getColumns();
+              LOG.debug("sourcing from columns: {}", columns);
+              conf.set(HBaseInputFormatGranular.COLUMN_LIST, columns);
+          }
+          break;
+
+          case REGIONAL:
+          {
+              conf.setInputFormat(HBaseInputFormatRegional.class);
+
+              String columns = getColumns();
+              LOG.debug("sourcing from columns: {}", columns);
+              conf.set(HBaseInputFormatRegional.COLUMN_LIST, columns);
+          }
+          break;
+
+          default:
+              LOG.error("Unknown Split Type : " + splitType.toString());
+      }
   }
 
   private String getColumns() {
@@ -345,4 +368,8 @@ public class HBaseScheme
     result = 31 * result + (valueFields != null ? Arrays.hashCode(valueFields) : 0);
     return result;
   }
+
+  public void setInputSplitTye(SplitType sType) {
+      this.splitType = sType;
+  }
 }
-- 
cgit v1.2.3