aboutsummaryrefslogtreecommitdiffstats
path: root/src/main/java/parallelai
diff options
context:
space:
mode:
Diffstat (limited to 'src/main/java/parallelai')
-rw-r--r--src/main/java/parallelai/spyglass/hbase/HBaseConstants.java1
-rw-r--r--src/main/java/parallelai/spyglass/hbase/HBaseOutputFormat.java3
-rw-r--r--src/main/java/parallelai/spyglass/hbase/HBaseRecordWriter.java3
-rw-r--r--src/main/java/parallelai/spyglass/hbase/HBaseTap.java10
4 files changed, 15 insertions, 2 deletions
diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseConstants.java b/src/main/java/parallelai/spyglass/hbase/HBaseConstants.java
index 7453d3e..a5e67b1 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseConstants.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseConstants.java
@@ -26,6 +26,7 @@ public class HBaseConstants {
public static final String VERSIONS = "hbase.%s.versions";
public static final String USE_SALT = "hbase.%s.use.salt";
public static final String SALT_PREFIX = "hbase.%s.salt.prefix";
+ public static final String AUTO_FLUSH = "hbase.%s.auto_flush";
public static final String SINK_MODE = "hbase.%s.sink.mode";
} \ No newline at end of file
diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseOutputFormat.java b/src/main/java/parallelai/spyglass/hbase/HBaseOutputFormat.java
index 401dea0..2ad8924 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseOutputFormat.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseOutputFormat.java
@@ -51,8 +51,7 @@ FileOutputFormat<ImmutableBytesWritable, Put> implements JobConfigurable {
LOG.error(e);
throw e;
}
- // TODO: Should Autoflush be set to true ???? - DONE
- table.setAutoFlush(true);
+ table.setAutoFlush(job.getBoolean(String.format(HBaseConstants.AUTO_FLUSH, tableName),false));
HBaseRecordWriter recordWriter = new HBaseRecordWriter(table);
recordWriter.setSinkMode(sinkMode);
return recordWriter;
diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseRecordWriter.java b/src/main/java/parallelai/spyglass/hbase/HBaseRecordWriter.java
index 50aa116..24654f5 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseRecordWriter.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseRecordWriter.java
@@ -30,6 +30,9 @@ public class HBaseRecordWriter
public void close(Reporter reporter)
throws IOException {
+ if (!m_table.isAutoFlush()) {
+ m_table.flushCommits();
+ }
m_table.close();
}
diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseTap.java b/src/main/java/parallelai/spyglass/hbase/HBaseTap.java
index 3576e96..e4a3f78 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseTap.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseTap.java
@@ -404,9 +404,11 @@ public class HBaseTap extends Tap<JobConf, RecordReader, OutputCollector> {
private static class SinkConfig implements Serializable {
public String tableName = null;
public boolean useSalt = false;
+ public boolean autoFlush = false;
public void configure(Configuration jobConf) {
jobConf.setBoolean(String.format(HBaseConstants.USE_SALT, tableName), useSalt);
+ jobConf.setBoolean(String.format(HBaseConstants.AUTO_FLUSH, tableName), autoFlush);
}
}
@@ -459,6 +461,14 @@ public class HBaseTap extends Tap<JobConf, RecordReader, OutputCollector> {
sinkConfigList.add(sc);
}
+
+ public void setHBaseSinkParms( boolean useSalt, boolean autoFlush ) {
+ SinkConfig sc = new SinkConfig();
+ sc.tableName = tableName;
+ sc.autoFlush = autoFlush;
+ sc.useSalt = useSalt;
+ sinkConfigList.add(sc);
+ }
private void setPrefixList(SourceConfig sc, String prefixList ) {
prefixList = (prefixList == null || prefixList.length() == 0) ? HBaseSalter.DEFAULT_PREFIX_LIST : prefixList;