diff --git a/sail/pom.xml b/sail/pom.xml
index ad2defbc2..234cd794b 100644
--- a/sail/pom.xml
+++ b/sail/pom.xml
@@ -61,13 +61,14 @@
 
         <dependency>
             <groupId>dev.langchain4j</groupId>
-            <artifactId>langchain4j-ollama</artifactId>
+            <artifactId>langchain4j-core</artifactId>
             <version>${langchain4j.version}</version>
         </dependency>
         <dependency>
             <groupId>dev.langchain4j</groupId>
-            <artifactId>langchain4j-local-ai</artifactId>
+            <artifactId>langchain4j-embeddings-all-minilm-l6-v2-q</artifactId>
             <version>${langchain4j.version}</version>
+            <scope>test</scope>
         </dependency>
 
         <dependency>
diff --git a/sail/src/main/java/com/msd/gin/halyard/sail/HBaseTripleSource.java b/sail/src/main/java/com/msd/gin/halyard/sail/HBaseTripleSource.java
index f5c7deed6..af78c2115 100644
--- a/sail/src/main/java/com/msd/gin/halyard/sail/HBaseTripleSource.java
+++ b/sail/src/main/java/com/msd/gin/halyard/sail/HBaseTripleSource.java
@@ -111,7 +111,7 @@ public final QueryPreparer newQueryPreparer() {
 	public final <T> T getQueryHelper(Class<T> qhType) {
 		Object qh = queryHelpers.get(qhType);
 		if (qh == null) {
-			throw new QueryEvaluationException(String.format("%s is not available", qhType.getName()));
+			throw new QueryEvaluationException(String.format("Query helper %s is not registered", qhType.getName()));
 		}
 		return qhType.cast(qh);
 	}
diff --git a/sail/src/test/java/com/msd/gin/halyard/sail/model/embedding/EmbeddingModelQueryHelperProviderTest.java b/sail/src/test/java/com/msd/gin/halyard/sail/model/embedding/EmbeddingModelQueryHelperProviderTest.java
index 514cae493..e32bd6e76 100644
--- a/sail/src/test/java/com/msd/gin/halyard/sail/model/embedding/EmbeddingModelQueryHelperProviderTest.java
+++ b/sail/src/test/java/com/msd/gin/halyard/sail/model/embedding/EmbeddingModelQueryHelperProviderTest.java
@@ -5,16 +5,15 @@
 
 import org.junit.jupiter.api.Test;
 
-import dev.langchain4j.model.localai.LocalAiEmbeddingModel;
+import dev.langchain4j.model.embedding.EmbeddingModel;
 
 public class EmbeddingModelQueryHelperProviderTest {
 	@Test
-	public void testLocalAI() throws Exception {
+	public void testProvider() throws Exception {
 		EmbeddingModelQueryHelperProvider provider = new EmbeddingModelQueryHelperProvider();
 		Map<String, String> config = new HashMap<>();
-		config.put("model.class", LocalAiEmbeddingModel.class.getName());
-		config.put("baseUrl", "http://localhost");
-		config.put("modelName", "llama3");
-		provider.createQueryHelper(config);
+		config.put("model.class", dev.langchain4j.model.embedding.onnx.allminilml6v2q.AllMiniLmL6V2QuantizedEmbeddingModel.class.getName());
+		EmbeddingModel model = provider.createQueryHelper(config);
+		model.embed("foobar");
 	}
 }
diff --git a/sdk/pom.xml b/sdk/pom.xml
index cda28bb40..28c865688 100644
--- a/sdk/pom.xml
+++ b/sdk/pom.xml
@@ -1,5 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<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">
+<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>
 	<artifactId>halyard-sdk</artifactId>
 	<packaging>pom</packaging>
@@ -28,21 +30,22 @@
 				</exclusion>
 			</exclusions>
 		</dependency>
-    <dependency>
-        <groupId>org.eclipse.rdf4j</groupId>
-        <artifactId>rdf4j-console</artifactId>
-        <version>${rdf4j.version}</version>
-        <exclusions>
+
+		<dependency>
+			<groupId>org.eclipse.rdf4j</groupId>
+			<artifactId>rdf4j-console</artifactId>
+			<version>${rdf4j.version}</version>
+			<exclusions>
 				<exclusion>
 					<groupId>org.eclipse.rdf4j</groupId>
 					<artifactId>rdf4j-spin</artifactId>
 				</exclusion>
-            <exclusion>
-                <groupId>ch.qos.logback</groupId>
-                <artifactId>logback-classic</artifactId>
-            </exclusion>
-        </exclusions>
-    </dependency>
+				<exclusion>
+					<groupId>ch.qos.logback</groupId>
+					<artifactId>logback-classic</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
 		<dependency>
 			<!-- for spark -->
 			<groupId>org.slf4j</groupId>
@@ -87,15 +90,23 @@
 						<phase>process-classes</phase>
 						<configuration>
 							<target>
-								<zip destfile="${project.build.directory}/rdf4j-repository-api-${rdf4j.version}.jar" update="true">
+								<zip
+									destfile="${project.build.directory}/rdf4j-repository-api-${rdf4j.version}.jar"
+									update="true">
 									<fileset dir="src/main/patches/rdf4j-client" />
 								</zip>
-								<get src="https://github.com/pulquero/hbase/releases/download/rel%2F${hbase.version}%2B${hadoop.version}/hbase-${hbase.version}-bin.tar.gz" dest="${project.build.directory}/hbase.tar.gz" skipexisting="true"/>
-								<untar src="${project.build.directory}/hbase.tar.gz" dest="${project.build.directory}/hbase-libs" compression="gzip">
+								<get
+									src="https://github.com/pulquero/hbase/releases/download/rel%2F${hbase.version}%2B${hadoop.version}/hbase-${hbase.version}-bin.tar.gz"
+									dest="${project.build.directory}/hbase.tar.gz"
+									skipexisting="true" />
+								<untar src="${project.build.directory}/hbase.tar.gz"
+									dest="${project.build.directory}/hbase-libs"
+									compression="gzip">
 									<patternset>
-										<include name="hbase-${hbase.version}/lib/hbase-*-${hbase.version}.jar"/>
+										<include
+											name="hbase-${hbase.version}/lib/hbase-*-${hbase.version}.jar" />
 									</patternset>
-									<mapper type="flatten"/>
+									<mapper type="flatten" />
 								</untar>
 							</target>
 						</configuration>
diff --git a/tools/pom.xml b/tools/pom.xml
index d32a9a016..800270d9b 100644
--- a/tools/pom.xml
+++ b/tools/pom.xml
@@ -86,6 +86,23 @@
 			<version>3.1</version>
 			<scope>runtime</scope>
 		</dependency>
+
+		<dependency>
+			<groupId>dev.langchain4j</groupId>
+			<artifactId>langchain4j-embeddings-all-minilm-l6-v2-q</artifactId>
+			<version>${langchain4j.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>dev.langchain4j</groupId>
+			<artifactId>langchain4j-ollama</artifactId>
+			<version>${langchain4j.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>dev.langchain4j</groupId>
+			<artifactId>langchain4j-local-ai</artifactId>
+			<version>${langchain4j.version}</version>
+		</dependency>
+
 		<dependency>
 			<groupId>${project.groupId}</groupId>
 			<artifactId>halyard-common</artifactId>
diff --git a/tools/src/main/java/com/msd/gin/halyard/tools/AbstractHalyardTool.java b/tools/src/main/java/com/msd/gin/halyard/tools/AbstractHalyardTool.java
index 457990ba9..5767ec43b 100644
--- a/tools/src/main/java/com/msd/gin/halyard/tools/AbstractHalyardTool.java
+++ b/tools/src/main/java/com/msd/gin/halyard/tools/AbstractHalyardTool.java
@@ -22,6 +22,7 @@
 import com.msd.gin.halyard.common.KeyspaceConnection;
 import com.msd.gin.halyard.common.RDFFactory;
 import com.msd.gin.halyard.common.StatementIndices;
+import com.msd.gin.halyard.rio.HRDFParser;
 import com.msd.gin.halyard.util.Version;
 
 import java.io.IOException;
@@ -51,6 +52,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
 import org.apache.hadoop.hbase.mapreduce.TableMapper;
 import org.apache.hadoop.hbase.regionserver.BloomType;
 import org.apache.hadoop.hbase.tool.BulkLoadHFiles;
@@ -64,9 +66,18 @@
 import org.eclipse.rdf4j.query.BindingSet;
 import org.eclipse.rdf4j.query.algebra.evaluation.QueryBindingSet;
 import org.eclipse.rdf4j.rio.helpers.NTriplesUtil;
+import org.eclipse.rdf4j.rio.nquads.NQuadsParserFactory;
+import org.eclipse.rdf4j.rio.ntriples.NTriplesParserFactory;
+import org.eclipse.rdf4j.rio.rdfjson.RDFJSONParserFactory;
+import org.eclipse.rdf4j.rio.rdfxml.RDFXMLParserFactory;
+import org.eclipse.rdf4j.rio.trig.TriGParserFactory;
+import org.eclipse.rdf4j.rio.trix.TriXParserFactory;
+import org.eclipse.rdf4j.rio.turtle.TurtleParserFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import dev.langchain4j.model.embedding.EmbeddingModel;
+
 /**
  *
  * @author Adam Sotona (MSD)
@@ -276,11 +287,36 @@ protected final List<Option> getRequiredOptions() {
         return requiredOptions;
     }
 
-    protected static final boolean isDryRun(Configuration conf) {
+    protected static boolean isDryRun(Configuration conf) {
     	return conf.getBoolean(DRY_RUN_PROPERTY, false);
     }
 
-    protected static final void bulkLoad(Job job, TableName tableName, Path workDir) throws IOException {
+    protected static void addRioDependencies(Configuration conf) throws IOException {
+    	TableMapReduceUtil.addDependencyJarsForClasses(conf,
+			TurtleParserFactory.class,
+			TriXParserFactory.class,
+			TriGParserFactory.class,
+			NTriplesParserFactory.class,
+			NQuadsParserFactory.class,
+			RDFXMLParserFactory.class,
+			RDFJSONParserFactory.class,
+			HRDFParser.Factory.class
+    	);
+    }
+
+    protected static void addLangModelDependencies(Configuration conf) throws IOException {
+    	TableMapReduceUtil.addDependencyJarsForClasses(conf,
+			EmbeddingModel.class,
+			dev.langchain4j.model.embedding.onnx.AbstractInProcessEmbeddingModel.class,
+			dev.langchain4j.model.embedding.onnx.allminilml6v2q.AllMiniLmL6V2QuantizedEmbeddingModel.class,
+			ai.onnxruntime.OrtEnvironment.class,
+			dev.langchain4j.model.localai.LocalAiEmbeddingModel.class,
+			dev.ai4j.openai4j.OpenAiClient.class,
+			dev.langchain4j.model.ollama.OllamaEmbeddingModel.class
+    	);
+    }
+
+    protected static void bulkLoad(Job job, TableName tableName, Path workDir) throws IOException {
     	// ensure job configuration is used
     	Configuration conf = job.getConfiguration();
     	if (isDryRun(conf)) {
@@ -292,7 +328,7 @@ protected static final void bulkLoad(Job job, TableName tableName, Path workDir)
     	}
     }
 
-    protected static final void addBloomFilterConfig(Configuration conf, TableName tableName) {
+    protected static void addBloomFilterConfig(Configuration conf, TableName tableName) {
 		byte[] tableAndFamily = HalyardTableUtils.getTableNameSuffixedWithFamily(tableName.toBytes());
 		Map<byte[], String> bloomTypeMap = createFamilyConfValueMap(conf, "hbase.hfileoutputformat.families.bloomtype");
 		String bloomType = bloomTypeMap.get(tableAndFamily);
diff --git a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkDelete.java b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkDelete.java
index 736a43c0f..73e232708 100644
--- a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkDelete.java
+++ b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkDelete.java
@@ -52,8 +52,6 @@
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
-import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
-import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.mapreduce.Job;
@@ -65,10 +63,6 @@
 import org.eclipse.rdf4j.model.Value;
 import org.eclipse.rdf4j.model.ValueFactory;
 import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
-import org.eclipse.rdf4j.rio.RDFFormat;
-import org.eclipse.rdf4j.rio.RDFParser;
-import org.eclipse.rdf4j.rio.Rio;
-import org.eclipse.rdf4j.rio.helpers.AbstractRDFHandler;
 import org.eclipse.rdf4j.rio.helpers.NTriplesUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -234,15 +228,6 @@ public int run(CommandLine cmd) throws Exception {
         configureBoolean(cmd, "dry-run");
         String snapshotPath = getConf().get(SNAPSHOT_PATH);
 
-        TableMapReduceUtil.addDependencyJarsForClasses(getConf(),
-            NTriplesUtil.class,
-            Rio.class,
-            AbstractRDFHandler.class,
-            RDFFormat.class,
-            RDFParser.class,
-            Table.class,
-            HBaseConfiguration.class,
-            AuthenticationProtos.class);
         HBaseConfiguration.addHbaseResources(getConf());
 
         RDFFactory rdfFactory;
diff --git a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkExport.java b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkExport.java
index 7c63d117a..18af51eb5 100644
--- a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkExport.java
+++ b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkExport.java
@@ -43,9 +43,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
-import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Job;
@@ -64,10 +62,6 @@
 import org.eclipse.rdf4j.query.algebra.TupleExpr;
 import org.eclipse.rdf4j.query.algebra.evaluation.QueryBindingSet;
 import org.eclipse.rdf4j.query.algebra.evaluation.QueryEvaluationStep;
-import org.eclipse.rdf4j.rio.RDFFormat;
-import org.eclipse.rdf4j.rio.RDFParser;
-import org.eclipse.rdf4j.rio.Rio;
-import org.eclipse.rdf4j.rio.helpers.AbstractRDFHandler;
 import org.eclipse.rdf4j.rio.helpers.NTriplesUtil;
 import org.elasticsearch.hadoop.mr.EsOutputFormat;
 import org.slf4j.Logger;
@@ -343,15 +337,8 @@ private static List<JsonInfo> run(Configuration conf, String queryFiles, String
             conf.set("es.input.json", "yes");
         }
 
-        TableMapReduceUtil.addDependencyJarsForClasses(conf,
-               NTriplesUtil.class,
-               Rio.class,
-               AbstractRDFHandler.class,
-               RDFFormat.class,
-               RDFParser.class,
-               Table.class,
-               HBaseConfiguration.class,
-               AuthenticationProtos.class);
+        addRioDependencies(conf);
+        addLangModelDependencies(conf);
         if (System.getProperty("exclude.es-hadoop") == null) {
          	TableMapReduceUtil.addDependencyJarsForClasses(conf, EsOutputFormat.class);
         }
diff --git a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkLoad.java b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkLoad.java
index 51ddaf96b..71db778e2 100644
--- a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkLoad.java
+++ b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkLoad.java
@@ -717,12 +717,7 @@ protected int run(CommandLine cmd) throws Exception {
 				tableDesc = hTable.getDescriptor();
 				RegionLocator regionLocator = conn.getRegionLocator(tableDesc.getTableName());
 				HFileOutputFormat2.configureIncrementalLoad(job, tableDesc, regionLocator);
-		        TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
-		                NTriplesUtil.class,
-		                Rio.class,
-		                AbstractRDFHandler.class,
-		                RDFFormat.class,
-		                RDFParser.class);
+		        addRioDependencies(job.getConfiguration());
 	        }
             try (Keyspace keyspace = HalyardTableUtils.getKeyspace(getConf(), conn, tableDesc.getTableName(), null, null)) {
             	try (KeyspaceConnection ksConn = keyspace.getConnection()) {
diff --git a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkUpdate.java b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkUpdate.java
index f3b36665b..59afff645 100644
--- a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkUpdate.java
+++ b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardBulkUpdate.java
@@ -52,8 +52,6 @@
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
-import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
-import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.Job;
@@ -76,10 +74,6 @@
 import org.eclipse.rdf4j.query.parser.ParsedUpdate;
 import org.eclipse.rdf4j.query.parser.QueryParserUtil;
 import org.eclipse.rdf4j.repository.sail.SailRepositoryConnection;
-import org.eclipse.rdf4j.rio.RDFFormat;
-import org.eclipse.rdf4j.rio.RDFParser;
-import org.eclipse.rdf4j.rio.Rio;
-import org.eclipse.rdf4j.rio.helpers.AbstractRDFHandler;
 import org.eclipse.rdf4j.rio.helpers.NTriplesUtil;
 import org.eclipse.rdf4j.sail.SailException;
 import org.slf4j.Logger;
@@ -353,15 +347,6 @@ static List<JsonInfo> executeUpdate(Configuration conf, String source, String qu
 
     private static List<JsonInfo> run(Configuration conf, String queryFiles, String query, String workdir) throws IOException, InterruptedException, ClassNotFoundException {
     	String source = conf.get(TABLE_NAME_PROPERTY);
-        TableMapReduceUtil.addDependencyJarsForClasses(conf,
-               NTriplesUtil.class,
-               Rio.class,
-               AbstractRDFHandler.class,
-               RDFFormat.class,
-               RDFParser.class,
-               Table.class,
-               HBaseConfiguration.class,
-               AuthenticationProtos.class);
         HBaseConfiguration.addHbaseResources(conf);
         // get bindings from merged configs
         BindingSet bindings = AbstractHalyardTool.getBindings(conf, SimpleValueFactory.getInstance());
diff --git a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardElasticIndexer.java b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardElasticIndexer.java
index fc22bd45b..56bb630ea 100644
--- a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardElasticIndexer.java
+++ b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardElasticIndexer.java
@@ -58,10 +58,8 @@
 import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
-import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
 import org.apache.hadoop.io.NullWritable;
@@ -76,11 +74,6 @@
 import org.eclipse.rdf4j.model.base.CoreDatatype.GEO;
 import org.eclipse.rdf4j.model.base.CoreDatatype.XSD;
 import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
-import org.eclipse.rdf4j.rio.RDFFormat;
-import org.eclipse.rdf4j.rio.RDFParser;
-import org.eclipse.rdf4j.rio.Rio;
-import org.eclipse.rdf4j.rio.helpers.AbstractRDFHandler;
-import org.eclipse.rdf4j.rio.helpers.NTriplesUtil;
 import org.elasticsearch.hadoop.mr.EsOutputFormat;
 import org.json.JSONArray;
 import org.json.JSONObject;
@@ -272,15 +265,6 @@ public int run(CommandLine cmd) throws Exception {
     		getConf().setBoolean(confProperty(TOOL_NAME, "fields."+field), true);
     	}
 
-        TableMapReduceUtil.addDependencyJarsForClasses(getConf(),
-                NTriplesUtil.class,
-                Rio.class,
-                AbstractRDFHandler.class,
-                RDFFormat.class,
-                RDFParser.class,
-                Table.class,
-                HBaseConfiguration.class,
-                AuthenticationProtos.class);
         if (System.getProperty("exclude.es-hadoop") == null) {
          	TableMapReduceUtil.addDependencyJarsForClasses(getConf(), EsOutputFormat.class);
         }
diff --git a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardPreSplit.java b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardPreSplit.java
index d49324aeb..ee312eaad 100644
--- a/tools/src/main/java/com/msd/gin/halyard/tools/HalyardPreSplit.java
+++ b/tools/src/main/java/com/msd/gin/halyard/tools/HalyardPreSplit.java
@@ -1,301 +1,291 @@
-/*
- * Copyright 2016 Merck Sharp & Dohme Corp. a subsidiary of Merck & Co.,
- * Inc., Kenilworth, NJ, USA.
- *
- * 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 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 com.msd.gin.halyard.tools;
-
-import static com.msd.gin.halyard.tools.HalyardBulkLoad.*;
-
-import com.msd.gin.halyard.common.HalyardTableUtils;
-import com.msd.gin.halyard.common.RDFFactory;
-import com.msd.gin.halyard.common.StatementIndices;
-import com.msd.gin.halyard.tools.HalyardBulkLoad.RioFileInputFormat;
-import com.msd.gin.halyard.util.LRUCache;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Set;
-import java.util.SplittableRandom;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.Admin;
-import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.client.ConnectionFactory;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
-import org.eclipse.rdf4j.model.Statement;
-import org.eclipse.rdf4j.rio.RDFFormat;
-import org.eclipse.rdf4j.rio.RDFParser;
-import org.eclipse.rdf4j.rio.Rio;
-import org.eclipse.rdf4j.rio.helpers.AbstractRDFHandler;
-import org.eclipse.rdf4j.rio.helpers.NTriplesUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Apache Hadoop MapReduce Tool for calculating pre-splits of an HBase table before a large dataset bulk-load.
- * Splits are based on the keys of a sample of the data to be loaded.
- * @author Adam Sotona (MSD)
- */
-public final class HalyardPreSplit extends AbstractHalyardTool {
-    private static final Logger LOG = LoggerFactory.getLogger(HalyardPreSplit.class);
-
-    private static final String TOOL_NAME = "presplit";
-    private static final String TABLE_PROPERTY = confProperty(TOOL_NAME, "table");
-    private static final String SPLIT_LIMIT_PROPERTY = confProperty(TOOL_NAME, "limit");
-    private static final String DECIMATION_FACTOR_PROPERTY = confProperty(TOOL_NAME, "decimation-factor");
-    private static final String OVERWRITE_PROPERTY = confProperty(TOOL_NAME, "overwrite");
-
-    private static final long DEFAULT_SPLIT_LIMIT = 55000000000l;
-    private static final int DEFAULT_DECIMATION_FACTOR = 1000;
-    private static final String NULL_TABLE = "-";
-
-    enum Counters {
-    	SAMPLED_STATEMENTS,
-    	TOTAL_STATEMENTS,
-    	TOTAL_STATEMENTS_READ,
-    	TOTAL_SPLITS
-    }
-
-    /**
-     * Mapper class transforming randomly selected sample of parsed Statement into set of HBase Keys and sizes
-     */
-    public final static class RDFDecimatingMapper extends Mapper<LongWritable, Statement, ImmutableBytesWritable, LongWritable> {
-
-        private final ImmutableBytesWritable rowKey = new ImmutableBytesWritable();
-        private final LongWritable keyValueLength = new LongWritable();
-        private final SplittableRandom random = new SplittableRandom();
-        private Set<Statement> stmtDedup;
-        private StatementIndices stmtIndices;
-        private int decimationFactor;
-        private long sampledStmts = 0L;
-        private long totalStmts = 0L;
-        private long totalStmtsRead = 0L;
-        private long nextStmtToSample = 0L;
-
-        @Override
-        protected void setup(Context context) throws IOException, InterruptedException {
-            Configuration conf = context.getConfiguration();
-            RDFFactory rdfFactory = RDFFactory.create(conf);
-            stmtDedup = Collections.newSetFromMap(new LRUCache<>(conf.getInt(STATEMENT_DEDUP_CACHE_SIZE_PROPERTY, DEFAULT_STATEMENT_DEDUP_CACHE_SIZE)));
-            stmtIndices = new StatementIndices(conf, rdfFactory);
-            decimationFactor = conf.getInt(DECIMATION_FACTOR_PROPERTY, DEFAULT_DECIMATION_FACTOR);
-        }
-
-        @Override
-        protected void map(LongWritable key, Statement stmt, final Context context) throws IOException, InterruptedException {
-        	// best effort statement deduplication
-        	if (stmtDedup.add(stmt)) {
-	        	if (totalStmts % decimationFactor == 0) {
-	        		// pick a random representative from the next decimationFactor worth of statements
-	       			nextStmtToSample = totalStmts + random.nextInt(decimationFactor);
-	        	}
-	        	if (totalStmts == nextStmtToSample) {
-	                List<? extends KeyValue> kvs = stmtIndices.insertKeyValues(stmt.getSubject(), stmt.getPredicate(), stmt.getObject(), stmt.getContext(), 0);
-	        		for (KeyValue keyValue: kvs) {
-	                    rowKey.set(keyValue.getRowArray(), keyValue.getRowOffset(), keyValue.getRowLength());
-	                    keyValueLength.set(keyValue.getLength());
-	                    context.write(rowKey, keyValueLength);
-	                }
-	        		sampledStmts++;
-	            }
-	            totalStmts++;
-        	}
-        	totalStmtsRead++;
-        }
-
-        @Override
-        protected void cleanup(Context context) throws IOException, InterruptedException {
-        	context.getCounter(Counters.SAMPLED_STATEMENTS).increment(sampledStmts);
-        	context.getCounter(Counters.TOTAL_STATEMENTS).increment(totalStmts);
-        	context.getCounter(Counters.TOTAL_STATEMENTS_READ).increment(totalStmtsRead);
-        }
-    }
-
-    static final class PreSplitReducer extends Reducer<ImmutableBytesWritable, LongWritable, NullWritable, NullWritable>  {
-    	private static final Logger logger = LoggerFactory.getLogger(PreSplitReducer.class);
-        private final List<byte[]> splits = new ArrayList<>();
-        private long splitLimit;
-        private int decimationFactor;
-        private byte lastRegion = 0;
-        private int keyCount = 0;
-        private int valueCount = 0;
-        private long valueSize = 0;
-        private int maxValueCount = 0;
-        private long maxValueSize = 0;
-        private byte[] maxCountKey;
-        private byte[] maxSizeKey;
-
-        @Override
-        protected void setup(Context context) throws IOException, InterruptedException {
-            splitLimit = context.getConfiguration().getLong(SPLIT_LIMIT_PROPERTY, DEFAULT_SPLIT_LIMIT);
-            decimationFactor = context.getConfiguration().getInt(DECIMATION_FACTOR_PROPERTY, DEFAULT_DECIMATION_FACTOR);
-            logger.info("NB: results may be affected by duplicate statements");
-        }
-
-        @Override
-        public void reduce(ImmutableBytesWritable key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
-            final byte region = key.get()[key.getOffset()];
-            boolean isNewRegion = (lastRegion != region);
-            final long splitSize = valueSize * decimationFactor;
-            if (isNewRegion || splitSize > splitLimit) {
-                byte[] splitBytes = isNewRegion ? new byte[]{region} : key.copyBytes();
-                splits.add(splitBytes);
-                int splitNum = splits.size();
-                String splitString = Bytes.toHex(splitBytes);
-                context.setStatus("#" + splitNum + " " + splitString);
-                logger.info("Split {} ({}): size {}, sampled keys {}, mean value count {}, mean value size {}, max value count {} ({}), max value size {} ({})", splitNum, splitString, splitSize, keyCount, valueCount/keyCount, valueSize/keyCount, maxValueCount, Bytes.toHex(maxCountKey), maxValueSize, Bytes.toHex(maxSizeKey));
-                lastRegion = region;
-                keyCount = 0;
-                valueCount = 0;
-                valueSize = 0;
-                maxValueCount = 0;
-                maxValueSize = 0;
-                maxCountKey = null;
-                maxSizeKey = null;
-            }
-            int sampleCount = 0;
-            int sampledSize = 0;
-            for (LongWritable val : values) {
-            	sampleCount++;
-            	sampledSize += val.get();
-            }
-            keyCount++;
-            valueCount += sampleCount;
-            valueSize += sampledSize;
-            if (sampleCount > maxValueCount) {
-            	maxValueCount = sampleCount;
-            	maxCountKey = key.copyBytes();
-            }
-            if (sampledSize > maxValueSize) {
-            	maxValueSize = sampledSize;
-            	maxSizeKey = key.copyBytes();
-            }
-        }
-
-        @Override
-        protected void cleanup(Context context) throws IOException, InterruptedException {
-        	context.getCounter(Counters.TOTAL_SPLITS).setValue(splits.size());
-            Configuration conf = context.getConfiguration();
-            String target = conf.get(TABLE_PROPERTY);
-            if (!NULL_TABLE.equals(target)) {
-	            TableName tableName = TableName.valueOf(target);
-	            boolean overwrite = conf.getBoolean(OVERWRITE_PROPERTY, false);
-	            try (Connection conn = HalyardTableUtils.getConnection(conf)) {
-	            	if (overwrite) {
-			    		try (Admin admin = conn.getAdmin()) {
-			    			if (admin.tableExists(tableName)) {
-			    				admin.disableTable(tableName);
-			    				admin.deleteTable(tableName);
-			    			}
-			    		}
-	            	}
-		            HalyardTableUtils.createTable(conn, tableName, splits.toArray(new byte[splits.size()][])).close();
-	            }
-            }
-        }
-    }
-
-    public HalyardPreSplit() {
-        super(
-            TOOL_NAME,
-            "Halyard Presplit is a MapReduce application designed to estimate optimal HBase region splits for big datasets before the Bulk Load. "
-                + "Halyard PreSplit creates an empty HBase table based on calculations from the dataset sources sampling. "
-                + "For very large datasets it is wise to calculate the pre-splits before the HBase table is created to allow more efficient following Bulk Load process of the data. "
-                + "Optional definition or override of the named graph should be specified exactly the same as for the following Bulk Load process so the region presplits estimations are precise.\n"
-                + "Halyard PreSplit consumes the same RDF data sources as Halyard Bulk Load.",
-            "Example: halyard presplit -s hdfs://my_RDF_files -t mydataset [-g 'http://whatever/graph']"
-        );
-        addOption("s", "source", "source_paths", SOURCE_PATHS_PROPERTY, "Source path(s) with RDF files, more paths can be delimited by comma, the paths are recursively searched for the supported files", true, true);
-        addOption("t", "target", "dataset_table", TABLE_PROPERTY, "Target HBase table with Halyard RDF store, optional HBase namespace of the target table must already exist", true, true);
-        addOption("i", "allow-invalid-iris", null, ALLOW_INVALID_IRIS_PROPERTY, "Optionally allow invalid IRI values (less overhead)", false, false);
-        addOption("k", "skip-invalid-lines", null, SKIP_INVALID_LINES_PROPERTY, "Optionally skip invalid lines", false, false);
-        addOption("g", "default-named-graph", "named_graph", DEFAULT_CONTEXT_PROPERTY, "Optionally specify default target named graph", false, true);
-        addOption("o", "named-graph-override", null, OVERRIDE_CONTEXT_PROPERTY, "Optionally override named graph also for quads, named graph is stripped from quads if --default-named-graph option is not specified", false, false);
-        addOption("d", "decimation-factor", "decimation_factor", DECIMATION_FACTOR_PROPERTY, String.format("Optionally overide pre-split random decimation factor (default is %d)", DEFAULT_DECIMATION_FACTOR), false, true);
-        addOption("l", "split-limit-splitSize", "splitSize", SPLIT_LIMIT_PROPERTY, String.format("Optionally override calculated split splitSize (default is %d)", DEFAULT_SPLIT_LIMIT), false, true);
-        addOption("f", "force", null, OVERWRITE_PROPERTY, "Overwrite existing table", false, false);
-    }
-
-    @Override
-    protected int run(CommandLine cmd) throws Exception {
-    	configureString(cmd, 's', null);
-    	configureString(cmd, 't', null);
-        configureBoolean(cmd, 'f');
-        String target = getConf().get(TABLE_PROPERTY);
-        if (!NULL_TABLE.equals(target) && !getConf().getBoolean(OVERWRITE_PROPERTY, false)) {
-	        try (Connection con = ConnectionFactory.createConnection(getConf())) {
-	            try (Admin admin = con.getAdmin()) {
-	                if (admin.tableExists(TableName.valueOf(target))) {
-	                    LOG.warn("Pre-split cannot modify already existing table {}", target);
-	                    return -1;
-	                }
-	            }
-	        }
-        }
-        configureBoolean(cmd, 'i');
-        configureBoolean(cmd, 'k');
-        configureIRIPattern(cmd, 'g', null);
-        configureBoolean(cmd, 'o');
-        configureInt(cmd, 'd', DEFAULT_DECIMATION_FACTOR, v -> {
-            if (v <= 0) {
-            	throw new IllegalArgumentException("Decimation factor must be greater than zero");
-            }
-        });
-        configureLong(cmd, 'l', DEFAULT_SPLIT_LIMIT);
-        String sourcePaths = getConf().get(SOURCE_PATHS_PROPERTY);
-        TableMapReduceUtil.addDependencyJarsForClasses(getConf(),
-                NTriplesUtil.class,
-                Rio.class,
-                AbstractRDFHandler.class,
-                RDFFormat.class,
-                RDFParser.class);
-        HBaseConfiguration.addHbaseResources(getConf());
-        Job job = Job.getInstance(getConf(), "HalyardPreSplit -> " + target);
-        job.setJarByClass(HalyardPreSplit.class);
-        job.setMapperClass(RDFDecimatingMapper.class);
-        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
-        job.setMapOutputValueClass(LongWritable.class);
-        job.setInputFormatClass(RioFileInputFormat.class);
-        FileInputFormat.setInputDirRecursive(job, true);
-        FileInputFormat.setInputPaths(job, sourcePaths);
-        TableMapReduceUtil.addDependencyJars(job);
-        TableMapReduceUtil.initCredentials(job);
-        job.setReducerClass(PreSplitReducer.class);
-        job.setNumReduceTasks(1);
-        job.setOutputFormatClass(NullOutputFormat.class);
-        if (job.waitForCompletion(true)) {
-            LOG.info("PreSplit Calculation completed.");
-            return 0;
-        } else {
-    		LOG.error("PreSplit failed to complete.");
-            return -1;
-        }
-    }
-}
+/*
+ * Copyright 2016 Merck Sharp & Dohme Corp. a subsidiary of Merck & Co.,
+ * Inc., Kenilworth, NJ, USA.
+ *
+ * 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 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 com.msd.gin.halyard.tools;
+
+import static com.msd.gin.halyard.tools.HalyardBulkLoad.*;
+
+import com.msd.gin.halyard.common.HalyardTableUtils;
+import com.msd.gin.halyard.common.RDFFactory;
+import com.msd.gin.halyard.common.StatementIndices;
+import com.msd.gin.halyard.tools.HalyardBulkLoad.RioFileInputFormat;
+import com.msd.gin.halyard.util.LRUCache;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.SplittableRandom;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
+import org.eclipse.rdf4j.model.Statement;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Apache Hadoop MapReduce Tool for calculating pre-splits of an HBase table before a large dataset bulk-load.
+ * Splits are based on the keys of a sample of the data to be loaded.
+ * @author Adam Sotona (MSD)
+ */
+public final class HalyardPreSplit extends AbstractHalyardTool {
+    private static final Logger LOG = LoggerFactory.getLogger(HalyardPreSplit.class);
+
+    private static final String TOOL_NAME = "presplit";
+    private static final String TABLE_PROPERTY = confProperty(TOOL_NAME, "table");
+    private static final String SPLIT_LIMIT_PROPERTY = confProperty(TOOL_NAME, "limit");
+    private static final String DECIMATION_FACTOR_PROPERTY = confProperty(TOOL_NAME, "decimation-factor");
+    private static final String OVERWRITE_PROPERTY = confProperty(TOOL_NAME, "overwrite");
+
+    private static final long DEFAULT_SPLIT_LIMIT = 55000000000l;
+    private static final int DEFAULT_DECIMATION_FACTOR = 1000;
+    private static final String NULL_TABLE = "-";
+
+    enum Counters {
+    	SAMPLED_STATEMENTS,
+    	TOTAL_STATEMENTS,
+    	TOTAL_STATEMENTS_READ,
+    	TOTAL_SPLITS
+    }
+
+    /**
+     * Mapper class transforming randomly selected sample of parsed Statement into set of HBase Keys and sizes
+     */
+    public final static class RDFDecimatingMapper extends Mapper<LongWritable, Statement, ImmutableBytesWritable, LongWritable> {
+
+        private final ImmutableBytesWritable rowKey = new ImmutableBytesWritable();
+        private final LongWritable keyValueLength = new LongWritable();
+        private final SplittableRandom random = new SplittableRandom();
+        private Set<Statement> stmtDedup;
+        private StatementIndices stmtIndices;
+        private int decimationFactor;
+        private long sampledStmts = 0L;
+        private long totalStmts = 0L;
+        private long totalStmtsRead = 0L;
+        private long nextStmtToSample = 0L;
+
+        @Override
+        protected void setup(Context context) throws IOException, InterruptedException {
+            Configuration conf = context.getConfiguration();
+            RDFFactory rdfFactory = RDFFactory.create(conf);
+            stmtDedup = Collections.newSetFromMap(new LRUCache<>(conf.getInt(STATEMENT_DEDUP_CACHE_SIZE_PROPERTY, DEFAULT_STATEMENT_DEDUP_CACHE_SIZE)));
+            stmtIndices = new StatementIndices(conf, rdfFactory);
+            decimationFactor = conf.getInt(DECIMATION_FACTOR_PROPERTY, DEFAULT_DECIMATION_FACTOR);
+        }
+
+        @Override
+        protected void map(LongWritable key, Statement stmt, final Context context) throws IOException, InterruptedException {
+        	// best effort statement deduplication
+        	if (stmtDedup.add(stmt)) {
+	        	if (totalStmts % decimationFactor == 0) {
+	        		// pick a random representative from the next decimationFactor worth of statements
+	       			nextStmtToSample = totalStmts + random.nextInt(decimationFactor);
+	        	}
+	        	if (totalStmts == nextStmtToSample) {
+	                List<? extends KeyValue> kvs = stmtIndices.insertKeyValues(stmt.getSubject(), stmt.getPredicate(), stmt.getObject(), stmt.getContext(), 0);
+	        		for (KeyValue keyValue: kvs) {
+	                    rowKey.set(keyValue.getRowArray(), keyValue.getRowOffset(), keyValue.getRowLength());
+	                    keyValueLength.set(keyValue.getLength());
+	                    context.write(rowKey, keyValueLength);
+	                }
+	        		sampledStmts++;
+	            }
+	            totalStmts++;
+        	}
+        	totalStmtsRead++;
+        }
+
+        @Override
+        protected void cleanup(Context context) throws IOException, InterruptedException {
+        	context.getCounter(Counters.SAMPLED_STATEMENTS).increment(sampledStmts);
+        	context.getCounter(Counters.TOTAL_STATEMENTS).increment(totalStmts);
+        	context.getCounter(Counters.TOTAL_STATEMENTS_READ).increment(totalStmtsRead);
+        }
+    }
+
+    static final class PreSplitReducer extends Reducer<ImmutableBytesWritable, LongWritable, NullWritable, NullWritable>  {
+    	private static final Logger logger = LoggerFactory.getLogger(PreSplitReducer.class);
+        private final List<byte[]> splits = new ArrayList<>();
+        private long splitLimit;
+        private int decimationFactor;
+        private byte lastRegion = 0;
+        private int keyCount = 0;
+        private int valueCount = 0;
+        private long valueSize = 0;
+        private int maxValueCount = 0;
+        private long maxValueSize = 0;
+        private byte[] maxCountKey;
+        private byte[] maxSizeKey;
+
+        @Override
+        protected void setup(Context context) throws IOException, InterruptedException {
+            splitLimit = context.getConfiguration().getLong(SPLIT_LIMIT_PROPERTY, DEFAULT_SPLIT_LIMIT);
+            decimationFactor = context.getConfiguration().getInt(DECIMATION_FACTOR_PROPERTY, DEFAULT_DECIMATION_FACTOR);
+            logger.info("NB: results may be affected by duplicate statements");
+        }
+
+        @Override
+        public void reduce(ImmutableBytesWritable key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
+            final byte region = key.get()[key.getOffset()];
+            boolean isNewRegion = (lastRegion != region);
+            final long splitSize = valueSize * decimationFactor;
+            if (isNewRegion || splitSize > splitLimit) {
+                byte[] splitBytes = isNewRegion ? new byte[]{region} : key.copyBytes();
+                splits.add(splitBytes);
+                int splitNum = splits.size();
+                String splitString = Bytes.toHex(splitBytes);
+                context.setStatus("#" + splitNum + " " + splitString);
+                logger.info("Split {} ({}): size {}, sampled keys {}, mean value count {}, mean value size {}, max value count {} ({}), max value size {} ({})", splitNum, splitString, splitSize, keyCount, valueCount/keyCount, valueSize/keyCount, maxValueCount, Bytes.toHex(maxCountKey), maxValueSize, Bytes.toHex(maxSizeKey));
+                lastRegion = region;
+                keyCount = 0;
+                valueCount = 0;
+                valueSize = 0;
+                maxValueCount = 0;
+                maxValueSize = 0;
+                maxCountKey = null;
+                maxSizeKey = null;
+            }
+            int sampleCount = 0;
+            int sampledSize = 0;
+            for (LongWritable val : values) {
+            	sampleCount++;
+            	sampledSize += val.get();
+            }
+            keyCount++;
+            valueCount += sampleCount;
+            valueSize += sampledSize;
+            if (sampleCount > maxValueCount) {
+            	maxValueCount = sampleCount;
+            	maxCountKey = key.copyBytes();
+            }
+            if (sampledSize > maxValueSize) {
+            	maxValueSize = sampledSize;
+            	maxSizeKey = key.copyBytes();
+            }
+        }
+
+        @Override
+        protected void cleanup(Context context) throws IOException, InterruptedException {
+        	context.getCounter(Counters.TOTAL_SPLITS).setValue(splits.size());
+            Configuration conf = context.getConfiguration();
+            String target = conf.get(TABLE_PROPERTY);
+            if (!NULL_TABLE.equals(target)) {
+	            TableName tableName = TableName.valueOf(target);
+	            boolean overwrite = conf.getBoolean(OVERWRITE_PROPERTY, false);
+	            try (Connection conn = HalyardTableUtils.getConnection(conf)) {
+	            	if (overwrite) {
+			    		try (Admin admin = conn.getAdmin()) {
+			    			if (admin.tableExists(tableName)) {
+			    				admin.disableTable(tableName);
+			    				admin.deleteTable(tableName);
+			    			}
+			    		}
+	            	}
+		            HalyardTableUtils.createTable(conn, tableName, splits.toArray(new byte[splits.size()][])).close();
+	            }
+            }
+        }
+    }
+
+    public HalyardPreSplit() {
+        super(
+            TOOL_NAME,
+            "Halyard Presplit is a MapReduce application designed to estimate optimal HBase region splits for big datasets before the Bulk Load. "
+                + "Halyard PreSplit creates an empty HBase table based on calculations from the dataset sources sampling. "
+                + "For very large datasets it is wise to calculate the pre-splits before the HBase table is created to allow more efficient following Bulk Load process of the data. "
+                + "Optional definition or override of the named graph should be specified exactly the same as for the following Bulk Load process so the region presplits estimations are precise.\n"
+                + "Halyard PreSplit consumes the same RDF data sources as Halyard Bulk Load.",
+            "Example: halyard presplit -s hdfs://my_RDF_files -t mydataset [-g 'http://whatever/graph']"
+        );
+        addOption("s", "source", "source_paths", SOURCE_PATHS_PROPERTY, "Source path(s) with RDF files, more paths can be delimited by comma, the paths are recursively searched for the supported files", true, true);
+        addOption("t", "target", "dataset_table", TABLE_PROPERTY, "Target HBase table with Halyard RDF store, optional HBase namespace of the target table must already exist", true, true);
+        addOption("i", "allow-invalid-iris", null, ALLOW_INVALID_IRIS_PROPERTY, "Optionally allow invalid IRI values (less overhead)", false, false);
+        addOption("k", "skip-invalid-lines", null, SKIP_INVALID_LINES_PROPERTY, "Optionally skip invalid lines", false, false);
+        addOption("g", "default-named-graph", "named_graph", DEFAULT_CONTEXT_PROPERTY, "Optionally specify default target named graph", false, true);
+        addOption("o", "named-graph-override", null, OVERRIDE_CONTEXT_PROPERTY, "Optionally override named graph also for quads, named graph is stripped from quads if --default-named-graph option is not specified", false, false);
+        addOption("d", "decimation-factor", "decimation_factor", DECIMATION_FACTOR_PROPERTY, String.format("Optionally overide pre-split random decimation factor (default is %d)", DEFAULT_DECIMATION_FACTOR), false, true);
+        addOption("l", "split-limit-splitSize", "splitSize", SPLIT_LIMIT_PROPERTY, String.format("Optionally override calculated split splitSize (default is %d)", DEFAULT_SPLIT_LIMIT), false, true);
+        addOption("f", "force", null, OVERWRITE_PROPERTY, "Overwrite existing table", false, false);
+    }
+
+    @Override
+    protected int run(CommandLine cmd) throws Exception {
+    	configureString(cmd, 's', null);
+    	configureString(cmd, 't', null);
+        configureBoolean(cmd, 'f');
+        String target = getConf().get(TABLE_PROPERTY);
+        if (!NULL_TABLE.equals(target) && !getConf().getBoolean(OVERWRITE_PROPERTY, false)) {
+	        try (Connection con = ConnectionFactory.createConnection(getConf())) {
+	            try (Admin admin = con.getAdmin()) {
+	                if (admin.tableExists(TableName.valueOf(target))) {
+	                    LOG.warn("Pre-split cannot modify already existing table {}", target);
+	                    return -1;
+	                }
+	            }
+	        }
+        }
+        configureBoolean(cmd, 'i');
+        configureBoolean(cmd, 'k');
+        configureIRIPattern(cmd, 'g', null);
+        configureBoolean(cmd, 'o');
+        configureInt(cmd, 'd', DEFAULT_DECIMATION_FACTOR, v -> {
+            if (v <= 0) {
+            	throw new IllegalArgumentException("Decimation factor must be greater than zero");
+            }
+        });
+        configureLong(cmd, 'l', DEFAULT_SPLIT_LIMIT);
+        String sourcePaths = getConf().get(SOURCE_PATHS_PROPERTY);
+        addRioDependencies(getConf());
+        HBaseConfiguration.addHbaseResources(getConf());
+        Job job = Job.getInstance(getConf(), "HalyardPreSplit -> " + target);
+        job.setJarByClass(HalyardPreSplit.class);
+        job.setMapperClass(RDFDecimatingMapper.class);
+        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
+        job.setMapOutputValueClass(LongWritable.class);
+        job.setInputFormatClass(RioFileInputFormat.class);
+        FileInputFormat.setInputDirRecursive(job, true);
+        FileInputFormat.setInputPaths(job, sourcePaths);
+        TableMapReduceUtil.addDependencyJars(job);
+        TableMapReduceUtil.initCredentials(job);
+        job.setReducerClass(PreSplitReducer.class);
+        job.setNumReduceTasks(1);
+        job.setOutputFormatClass(NullOutputFormat.class);
+        if (job.waitForCompletion(true)) {
+            LOG.info("PreSplit Calculation completed.");
+            return 0;
+        } else {
+    		LOG.error("PreSplit failed to complete.");
+            return -1;
+        }
+    }
+}