http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/BulkNtripsInputTool.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/BulkNtripsInputTool.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/BulkNtripsInputTool.java new file mode 100644 index 0000000..8f6184b --- /dev/null +++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/BulkNtripsInputTool.java @@ -0,0 +1,368 @@ +package mvm.rya.accumulo.mr.fileinput; + +/* + * #%L + * mvm.rya.accumulo.rya + * %% + * Copyright (C) 2014 Rya + * %% + * 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. + * #L% + */ + +import static com.google.common.base.Preconditions.checkNotNull; +import static mvm.rya.accumulo.AccumuloRdfUtils.extractValue; +import static mvm.rya.accumulo.AccumuloRdfUtils.from; + +import java.io.BufferedOutputStream; +import java.io.IOException; +import java.io.PrintStream; +import java.io.StringReader; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Map; + +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.accumulo.AccumuloRdfConstants; +import mvm.rya.accumulo.mr.utils.MRUtils; +import mvm.rya.api.RdfCloudTripleStoreConstants; +import mvm.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.domain.RyaURI; +import mvm.rya.api.resolver.RdfToRyaConversions; +import mvm.rya.api.resolver.RyaTripleContext; +import mvm.rya.api.resolver.triple.TripleRow; +import mvm.rya.api.resolver.triple.TripleRowResolver; + +import org.apache.accumulo.core.client.Connector; +import org.apache.accumulo.core.client.ZooKeeperInstance; +import org.apache.accumulo.core.client.admin.TableOperations; +import org.apache.accumulo.core.client.mapreduce.AccumuloFileOutputFormat; +import org.apache.accumulo.core.client.mapreduce.lib.partition.KeyRangePartitioner; +import org.apache.accumulo.core.client.mapreduce.lib.partition.RangePartitioner; +import org.apache.accumulo.core.client.security.tokens.PasswordToken; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.util.TextUtil; +import org.apache.commons.codec.binary.Base64; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.conf.Configured; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.permission.FsAction; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.Text; +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.TextInputFormat; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; +import org.openrdf.model.Statement; +import org.openrdf.rio.ParserConfig; +import org.openrdf.rio.RDFFormat; +import org.openrdf.rio.RDFHandler; +import org.openrdf.rio.RDFHandlerException; +import org.openrdf.rio.RDFParseException; +import org.openrdf.rio.RDFParser; +import org.openrdf.rio.nquads.NQuadsParser; + +import com.google.common.base.Preconditions; + +/** + * Take large ntrips files and use MapReduce and Cloudbase + * Bulk ingest techniques to load into the table in our partition format. + * <p/> + * Input: NTrips file + * Map: + * - key : shard row - Text + * - value : stmt in doc triple format - Text + * Partitioner: RangePartitioner + * Reduce: + * - key : all the entries for each triple - Cloudbase Key + * Class BulkNtripsInputTool + * Date: Sep 13, 2011 + * Time: 10:00:17 AM + */ +public class BulkNtripsInputTool extends Configured implements Tool { + + public static final String WORKDIR_PROP = "bulk.n3.workdir"; + + private String userName = "root"; + private String pwd = "root"; + private String instance = "isntance"; + private String zk = "zoo"; + private String ttl = null; + private String workDirBase = "/temp/bulkcb/work"; + private String format = RDFFormat.NQUADS.getName(); + + @Override + public int run(final String[] args) throws Exception { + final Configuration conf = getConf(); + try { + //conf + zk = conf.get(MRUtils.AC_ZK_PROP, zk); + ttl = conf.get(MRUtils.AC_TTL_PROP, ttl); + instance = conf.get(MRUtils.AC_INSTANCE_PROP, instance); + userName = conf.get(MRUtils.AC_USERNAME_PROP, userName); + pwd = conf.get(MRUtils.AC_PWD_PROP, pwd); + workDirBase = conf.get(WORKDIR_PROP, workDirBase); + format = conf.get(MRUtils.FORMAT_PROP, format); + conf.set(MRUtils.FORMAT_PROP, format); + final String inputDir = args[0]; + + ZooKeeperInstance zooKeeperInstance = new ZooKeeperInstance(instance, zk); + Connector connector = zooKeeperInstance.getConnector(userName, new PasswordToken(pwd)); + TableOperations tableOperations = connector.tableOperations(); + + if (conf.get(AccumuloRdfConfiguration.CONF_ADDITIONAL_INDEXERS) != null ) { + throw new IllegalArgumentException("Cannot use Bulk N Trips tool with Additional Indexers"); + } + + String tablePrefix = conf.get(MRUtils.TABLE_PREFIX_PROPERTY, null); + if (tablePrefix != null) + RdfCloudTripleStoreConstants.prefixTables(tablePrefix); + String[] tables = {tablePrefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX, + tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, + tablePrefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX}; + Collection<Job> jobs = new ArrayList<Job>(); + for (final String tableName : tables) { + PrintStream out = null; + try { + String workDir = workDirBase + "/" + tableName; + System.out.println("Loading data into table[" + tableName + "]"); + + Job job = new Job(new Configuration(conf), "Bulk Ingest load data to Generic RDF Table[" + tableName + "]"); + job.setJarByClass(this.getClass()); + //setting long job + Configuration jobConf = job.getConfiguration(); + jobConf.setBoolean("mapred.map.tasks.speculative.execution", false); + jobConf.setBoolean("mapred.reduce.tasks.speculative.execution", false); + jobConf.set("io.sort.mb", jobConf.get("io.sort.mb", "256")); + jobConf.setBoolean("mapred.compress.map.output", true); +// jobConf.set("mapred.map.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec"); //TODO: I would like LZO compression + + job.setInputFormatClass(TextInputFormat.class); + + job.setMapperClass(ParseNtripsMapper.class); + job.setMapOutputKeyClass(Key.class); + job.setMapOutputValueClass(Value.class); + + job.setCombinerClass(OutStmtMutationsReducer.class); + job.setReducerClass(OutStmtMutationsReducer.class); + job.setOutputFormatClass(AccumuloFileOutputFormat.class); + // AccumuloFileOutputFormat.setZooKeeperInstance(jobConf, instance, zk); + + jobConf.set(ParseNtripsMapper.TABLE_PROPERTY, tableName); + + TextInputFormat.setInputPaths(job, new Path(inputDir)); + + FileSystem fs = FileSystem.get(conf); + Path workPath = new Path(workDir); + if (fs.exists(workPath)) + fs.delete(workPath, true); + + //make failures dir + Path failures = new Path(workDir, "failures"); + fs.delete(failures, true); + fs.mkdirs(new Path(workDir, "failures")); + + AccumuloFileOutputFormat.setOutputPath(job, new Path(workDir + "/files")); + + out = new PrintStream(new BufferedOutputStream(fs.create(new Path(workDir + "/splits.txt")))); + + if (!tableOperations.exists(tableName)) + tableOperations.create(tableName); + Collection<Text> splits = tableOperations.getSplits(tableName, Integer.MAX_VALUE); + for (Text split : splits) + out.println(new String(Base64.encodeBase64(TextUtil.getBytes(split)))); + + job.setNumReduceTasks(splits.size() + 1); + out.close(); + + job.setPartitionerClass(KeyRangePartitioner.class); + RangePartitioner.setSplitFile(job, workDir + "/splits.txt"); + + jobConf.set(WORKDIR_PROP, workDir); + + job.submit(); + jobs.add(job); + + } catch (Exception re) { + throw new RuntimeException(re); + } finally { + if (out != null) + out.close(); + } + } + + for (Job job : jobs) { + while (!job.isComplete()) { + Thread.sleep(1000); + } + } + + for(String tableName : tables) { + String workDir = workDirBase + "/" + tableName; + String filesDir = workDir + "/files"; + String failuresDir = workDir + "/failures"; + + FileSystem fs = FileSystem.get(conf); + + //make sure that the "accumulo" user can read/write/execute into these directories this path + fs.setPermission(new Path(filesDir), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL)); + fs.setPermission(new Path(failuresDir), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL)); + + tableOperations.importDirectory( + tableName, + filesDir, + failuresDir, + false); + + } + + } catch (Exception e ){ + throw new RuntimeException(e); + } + + return 0; + } + + public static void main(String[] args) throws Exception { + ToolRunner.run(new Configuration(), new BulkNtripsInputTool(), args); + } + + /** + * input: ntrips format triple + * <p/> + * output: key: shard row from generator + * value: stmt in serialized format (document format) + */ + public static class ParseNtripsMapper extends Mapper<LongWritable, Text, Key, Value> { + public static final String TABLE_PROPERTY = "parsentripsmapper.table"; + + private RDFParser parser; + private String rdfFormat; + private String namedGraph; + private RyaTripleContext ryaContext; + private TripleRowResolver rowResolver; + + @Override + protected void setup(final Context context) throws IOException, InterruptedException { + super.setup(context); + Configuration conf = context.getConfiguration(); + final String table = conf.get(TABLE_PROPERTY); + Preconditions.checkNotNull(table, "Set the " + TABLE_PROPERTY + " property on the map reduce job"); + this.ryaContext = RyaTripleContext.getInstance(new AccumuloRdfConfiguration(conf)); + rowResolver = ryaContext.getTripleResolver(); + + final String cv_s = conf.get(MRUtils.AC_CV_PROP); + final byte[] cv = cv_s == null ? null : cv_s.getBytes(); + rdfFormat = conf.get(MRUtils.FORMAT_PROP); + checkNotNull(rdfFormat, "Rdf format cannot be null"); + + namedGraph = conf.get(MRUtils.NAMED_GRAPH_PROP); + + parser = new NQuadsParser(); + parser.setParserConfig(new ParserConfig(true, true, true, RDFParser.DatatypeHandling.VERIFY)); + parser.setRDFHandler(new RDFHandler() { + + @Override + public void startRDF() throws RDFHandlerException { + + } + + @Override + public void endRDF() throws RDFHandlerException { + + } + + @Override + public void handleNamespace(String s, String s1) throws RDFHandlerException { + + } + + @Override + public void handleStatement(Statement statement) throws RDFHandlerException { + try { + RyaStatement rs = RdfToRyaConversions.convertStatement(statement); + if(rs.getColumnVisibility() == null) { + rs.setColumnVisibility(cv); + } + + // Inject the specified context into the statement. + if(namedGraph != null){ + rs.setContext(new RyaURI(namedGraph)); + } else if (statement.getContext() != null) { + rs.setContext(new RyaURI(statement.getContext().toString())); + } + + Map<RdfCloudTripleStoreConstants.TABLE_LAYOUT,TripleRow> serialize = rowResolver.serialize(rs); + + if (table.contains(RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX)) { + TripleRow tripleRow = serialize.get(TABLE_LAYOUT.SPO); + context.write( + from(tripleRow), + extractValue(tripleRow) + ); + } else if (table.contains(RdfCloudTripleStoreConstants.TBL_PO_SUFFIX)) { + TripleRow tripleRow = serialize.get(TABLE_LAYOUT.PO); + context.write( + from(tripleRow), + extractValue(tripleRow) + ); + } else if (table.contains(RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX)) { + TripleRow tripleRow = serialize.get(TABLE_LAYOUT.OSP); + context.write( + from(tripleRow), + extractValue(tripleRow) + ); + } else + throw new IllegalArgumentException("Unrecognized table[" + table + "]"); + + } catch (Exception e) { + throw new RDFHandlerException(e); + } + } + + @Override + public void handleComment(String s) throws RDFHandlerException { + + } + }); + } + + @Override + public void map(LongWritable key, Text value, Context output) + throws IOException, InterruptedException { + String rdf = value.toString(); + try { + parser.parse(new StringReader(rdf), ""); + } catch (RDFParseException e) { + System.out.println("Line[" + rdf + "] cannot be formatted with format[" + rdfFormat + "]. Exception[" + e.getMessage() + "]"); + } catch (Exception e) { + e.printStackTrace(); + throw new IOException("Exception occurred parsing triple[" + rdf + "]"); + } + } + } + + public static class OutStmtMutationsReducer extends Reducer<Key, Value, Key, Value> { + + public void reduce(Key key, Iterable<Value> values, Context output) + throws IOException, InterruptedException { + output.write(key, AccumuloRdfConstants.EMPTY_VALUE); + } + } +}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputByLineTool.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputByLineTool.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputByLineTool.java new file mode 100644 index 0000000..7916cdb --- /dev/null +++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputByLineTool.java @@ -0,0 +1,250 @@ +package mvm.rya.accumulo.mr.fileinput; + +/* + * #%L + * mvm.rya.accumulo.rya + * %% + * Copyright (C) 2014 Rya + * %% + * 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. + * #L% + */ + +import java.io.IOException; +import java.io.StringReader; +import java.util.Collection; +import java.util.Date; +import java.util.Map; + +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.accumulo.AccumuloRdfConstants; +import mvm.rya.accumulo.RyaTableMutationsFactory; +import mvm.rya.accumulo.mr.utils.MRUtils; +import mvm.rya.api.RdfCloudTripleStoreConstants; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.resolver.RdfToRyaConversions; +import mvm.rya.api.resolver.RyaTripleContext; + +import org.apache.accumulo.core.client.AccumuloSecurityException; +import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat; +import org.apache.accumulo.core.client.security.tokens.PasswordToken; +import org.apache.accumulo.core.data.Mutation; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.Mapper; +import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; +import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; +import org.openrdf.model.Statement; +import org.openrdf.rio.RDFFormat; +import org.openrdf.rio.RDFHandler; +import org.openrdf.rio.RDFHandlerException; +import org.openrdf.rio.RDFParser; +import org.openrdf.rio.Rio; + +/** + * Do bulk import of rdf files + * Class RdfFileInputTool2 + * Date: May 16, 2011 + * Time: 3:12:16 PM + */ +public class RdfFileInputByLineTool implements Tool { + + private Configuration conf = new Configuration(); + + private String userName = "root"; + private String pwd = "password"; + private String instance = "instance"; + private String zk = "zoo"; + private String tablePrefix = null; + private RDFFormat format = RDFFormat.NTRIPLES; + + public Configuration getConf() { + return conf; + } + + public void setConf(Configuration conf) { + this.conf = conf; + } + + public static void main(String[] args) { + try { + ToolRunner.run(new Configuration(), new RdfFileInputByLineTool(), args); + } catch (Exception e) { + e.printStackTrace(); + } + } + + public long runJob(String[] args) throws IOException, ClassNotFoundException, InterruptedException, AccumuloSecurityException { + conf.setBoolean("mapred.map.tasks.speculative.execution", false); + conf.setBoolean("mapred.reduce.tasks.speculative.execution", false); + conf.set("io.sort.mb", "256"); + conf.setLong("mapred.task.timeout", 600000000); + + zk = conf.get(MRUtils.AC_ZK_PROP, zk); + instance = conf.get(MRUtils.AC_INSTANCE_PROP, instance); + userName = conf.get(MRUtils.AC_USERNAME_PROP, userName); + pwd = conf.get(MRUtils.AC_PWD_PROP, pwd); + format = RDFFormat.valueOf(conf.get(MRUtils.FORMAT_PROP, RDFFormat.NTRIPLES.toString())); + + String tablePrefix = conf.get(MRUtils.TABLE_PREFIX_PROPERTY, RdfCloudTripleStoreConstants.TBL_PRFX_DEF); + + Job job = new Job(conf); + job.setJarByClass(RdfFileInputByLineTool.class); + + // set up cloudbase input + job.setInputFormatClass(TextInputFormat.class); + FileInputFormat.addInputPath(job, new Path(args[0])); + + // set input output of the particular job + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(Mutation.class); + + job.setOutputFormatClass(AccumuloOutputFormat.class); + AccumuloOutputFormat.setConnectorInfo(job, userName, new PasswordToken(pwd.getBytes())); + AccumuloOutputFormat.setCreateTables(job, true); + AccumuloOutputFormat.setDefaultTableName(job, tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX); + AccumuloOutputFormat.setZooKeeperInstance(job, instance, zk); + + // set mapper and reducer classes + job.setMapperClass(TextToMutationMapper.class); + job.setNumReduceTasks(0); + + // Submit the job + Date startTime = new Date(); + System.out.println("Job started: " + startTime); + int exitCode = job.waitForCompletion(true) ? 0 : 1; + + if (exitCode == 0) { + Date end_time = new Date(); + System.out.println("Job ended: " + end_time); + System.out.println("The job took " + + (end_time.getTime() - startTime.getTime()) / 1000 + + " seconds."); + return job + .getCounters() + .findCounter("org.apache.hadoop.mapred.Task$Counter", + "REDUCE_OUTPUT_RECORDS").getValue(); + } else { + System.out.println("Job Failed!!!"); + } + + return -1; + } + + @Override + public int run(String[] args) throws Exception { + return (int) runJob(args); + } + + public static class TextToMutationMapper extends Mapper<LongWritable, Text, Text, Mutation> { + protected RDFParser parser; + private String prefix; + private RDFFormat rdfFormat; + protected Text spo_table; + private Text po_table; + private Text osp_table; + private byte[] cv = AccumuloRdfConstants.EMPTY_CV.getExpression(); + + public TextToMutationMapper() { + } + + @Override + protected void setup(final Context context) throws IOException, InterruptedException { + super.setup(context); + Configuration conf = context.getConfiguration(); + prefix = conf.get(MRUtils.TABLE_PREFIX_PROPERTY, null); + if (prefix != null) { + RdfCloudTripleStoreConstants.prefixTables(prefix); + } + + spo_table = new Text(prefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX); + po_table = new Text(prefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX); + osp_table = new Text(prefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX); + + final String cv_s = conf.get(MRUtils.AC_CV_PROP); + if (cv_s != null) + cv = cv_s.getBytes(); + + rdfFormat = RDFFormat.valueOf(conf.get(MRUtils.FORMAT_PROP, RDFFormat.NTRIPLES.toString())); + parser = Rio.createParser(rdfFormat); + RyaTripleContext tripleContext = RyaTripleContext.getInstance(new AccumuloRdfConfiguration(conf)); + final RyaTableMutationsFactory mut = new RyaTableMutationsFactory(tripleContext); + + parser.setRDFHandler(new RDFHandler() { + + @Override + public void startRDF() throws RDFHandlerException { + + } + + @Override + public void endRDF() throws RDFHandlerException { + + } + + @Override + public void handleNamespace(String s, String s1) throws RDFHandlerException { + + } + + @Override + public void handleStatement(Statement statement) throws RDFHandlerException { + try { + RyaStatement ryaStatement = RdfToRyaConversions.convertStatement(statement); + if(ryaStatement.getColumnVisibility() == null) { + ryaStatement.setColumnVisibility(cv); + } + Map<RdfCloudTripleStoreConstants.TABLE_LAYOUT, Collection<Mutation>> mutationMap = + mut.serialize(ryaStatement); + Collection<Mutation> spo = mutationMap.get(RdfCloudTripleStoreConstants.TABLE_LAYOUT.SPO); + Collection<Mutation> po = mutationMap.get(RdfCloudTripleStoreConstants.TABLE_LAYOUT.PO); + Collection<Mutation> osp = mutationMap.get(RdfCloudTripleStoreConstants.TABLE_LAYOUT.OSP); + + for (Mutation m : spo) { + context.write(spo_table, m); + } + for (Mutation m : po) { + context.write(po_table, m); + } + for (Mutation m : osp) { + context.write(osp_table, m); + } + } catch (Exception e) { + throw new RDFHandlerException(e); + } + } + + @Override + public void handleComment(String s) throws RDFHandlerException { + + } + }); + } + + @Override + protected void map(LongWritable key, Text value, final Context context) throws IOException, InterruptedException { + try { + parser.parse(new StringReader(value.toString()), ""); + } catch (Exception e) { + throw new IOException(e); + } + } + + } +} + http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputFormat.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputFormat.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputFormat.java new file mode 100644 index 0000000..ba88a7e --- /dev/null +++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputFormat.java @@ -0,0 +1,145 @@ +package mvm.rya.accumulo.mr.fileinput; + +/* + * #%L + * mvm.rya.accumulo.rya + * %% + * Copyright (C) 2014 Rya + * %% + * 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. + * #L% + */ + +import java.io.IOException; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.accumulo.mr.utils.MRUtils; +import mvm.rya.api.domain.utils.RyaStatementWritable; +import mvm.rya.api.resolver.RdfToRyaConversions; +import mvm.rya.api.resolver.RyaTripleContext; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.hadoop.mapreduce.RecordReader; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; +import org.apache.hadoop.mapreduce.lib.input.FileSplit; +import org.openrdf.model.Statement; +import org.openrdf.rio.RDFFormat; +import org.openrdf.rio.RDFHandler; +import org.openrdf.rio.RDFHandlerException; +import org.openrdf.rio.RDFParser; +import org.openrdf.rio.Rio; + +/** + * Be able to input multiple rdf formatted files. Convert from rdf format to statements. + * Class RdfFileInputFormat + * Date: May 16, 2011 + * Time: 2:11:24 PM + */ +public class RdfFileInputFormat extends FileInputFormat<LongWritable, RyaStatementWritable> { + + @Override + public RecordReader<LongWritable, RyaStatementWritable> createRecordReader(InputSplit inputSplit, + TaskAttemptContext taskAttemptContext) + throws IOException, InterruptedException { + return new RdfFileRecordReader(); + } + + private class RdfFileRecordReader extends RecordReader<LongWritable, RyaStatementWritable> implements RDFHandler { + + boolean closed = false; + long count = 0; + BlockingQueue<RyaStatementWritable> queue = new LinkedBlockingQueue<RyaStatementWritable>(); + int total = 0; + private RyaTripleContext tripleContext; + + + @Override + public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { + FileSplit fileSplit = (FileSplit) inputSplit; + Configuration conf = taskAttemptContext.getConfiguration(); + String rdfForm_s = conf.get(MRUtils.FORMAT_PROP, RDFFormat.RDFXML.getName()); + RDFFormat rdfFormat = RDFFormat.valueOf(rdfForm_s); + tripleContext = RyaTripleContext.getInstance(new AccumuloRdfConfiguration(conf)); + + Path file = fileSplit.getPath(); + FileSystem fs = file.getFileSystem(conf); + FSDataInputStream fileIn = fs.open(fileSplit.getPath()); + + RDFParser rdfParser = Rio.createParser(rdfFormat); + rdfParser.setRDFHandler(this); + try { + rdfParser.parse(fileIn, ""); + } catch (Exception e) { + throw new IOException(e); + } + fileIn.close(); + total = queue.size(); + //TODO: Make this threaded so that you don't hold too many statements before sending them + } + + @Override + public boolean nextKeyValue() throws IOException, InterruptedException { + return queue.size() > 0; + } + + @Override + public LongWritable getCurrentKey() throws IOException, InterruptedException { + return new LongWritable(count++); + } + + @Override + public RyaStatementWritable getCurrentValue() throws IOException, InterruptedException { + return queue.poll(); + } + + @Override + public float getProgress() throws IOException, InterruptedException { + return ((float) (total - queue.size())) / ((float) total); + } + + @Override + public void close() throws IOException { + closed = true; + } + + @Override + public void startRDF() throws RDFHandlerException { + } + + @Override + public void endRDF() throws RDFHandlerException { + } + + @Override + public void handleNamespace(String s, String s1) throws RDFHandlerException { + } + + @Override + public void handleStatement(Statement statement) throws RDFHandlerException { + queue.add(new RyaStatementWritable(RdfToRyaConversions.convertStatement(statement), tripleContext)); + } + + @Override + public void handleComment(String s) throws RDFHandlerException { + } + } + +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputTool.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputTool.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputTool.java new file mode 100644 index 0000000..789ea88 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputTool.java @@ -0,0 +1,174 @@ +package mvm.rya.accumulo.mr.fileinput; + +/* + * #%L + * mvm.rya.accumulo.rya + * %% + * Copyright (C) 2014 Rya + * %% + * 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. + * #L% + */ + +import static mvm.rya.accumulo.AccumuloRdfConstants.EMPTY_CV; + +import java.io.IOException; +import java.util.Collection; +import java.util.Date; +import java.util.Map; + +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.accumulo.RyaTableMutationsFactory; +import mvm.rya.accumulo.mr.AbstractAccumuloMRTool; +import mvm.rya.accumulo.mr.utils.MRUtils; +import mvm.rya.api.RdfCloudTripleStoreConstants; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.domain.utils.RyaStatementWritable; +import mvm.rya.api.resolver.RyaTripleContext; + +import org.apache.accumulo.core.client.AccumuloSecurityException; +import org.apache.accumulo.core.data.Mutation; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.Mapper; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; +import org.openrdf.rio.RDFFormat; + +/** + * Do bulk import of rdf files + * Class RdfFileInputTool + * Date: May 16, 2011 + * Time: 3:12:16 PM + */ +public class RdfFileInputTool extends AbstractAccumuloMRTool implements Tool { + + private String format = RDFFormat.RDFXML.getName(); + + public static void main(String[] args) { + try { + ToolRunner.run(new Configuration(), new RdfFileInputTool(), args); + } catch (Exception e) { + e.printStackTrace(); + } + } + + public long runJob(String[] args) throws IOException, ClassNotFoundException, InterruptedException, AccumuloSecurityException { + conf.set(MRUtils.JOB_NAME_PROP, "Rdf File Input"); + //faster + init(); + format = conf.get(MRUtils.FORMAT_PROP, format); + conf.set(MRUtils.FORMAT_PROP, format); + + String inputPath = conf.get(MRUtils.INPUT_PATH, args[0]); + + Job job = new Job(conf); + job.setJarByClass(RdfFileInputTool.class); + + // set up cloudbase input + job.setInputFormatClass(RdfFileInputFormat.class); + RdfFileInputFormat.addInputPath(job, new Path(inputPath)); + + // set input output of the particular job + job.setMapOutputKeyClass(LongWritable.class); + job.setMapOutputValueClass(RyaStatementWritable.class); + + setupOutputFormat(job, tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX); + + // set mapper and reducer classes + job.setMapperClass(StatementToMutationMapper.class); + job.setNumReduceTasks(0); + + // Submit the job + Date startTime = new Date(); + System.out.println("Job started: " + startTime); + int exitCode = job.waitForCompletion(true) ? 0 : 1; + + if (exitCode == 0) { + Date end_time = new Date(); + System.out.println("Job ended: " + end_time); + System.out.println("The job took " + + (end_time.getTime() - startTime.getTime()) / 1000 + + " seconds."); + return job + .getCounters() + .findCounter("org.apache.hadoop.mapred.Task$Counter", + "REDUCE_OUTPUT_RECORDS").getValue(); + } else { + System.out.println("Job Failed!!!"); + } + + return -1; + } + + @Override + public int run(String[] args) throws Exception { + runJob(args); + return 0; + } + + public static class StatementToMutationMapper extends Mapper<LongWritable, RyaStatementWritable, Text, Mutation> { + protected String tablePrefix; + protected Text spo_table; + protected Text po_table; + protected Text osp_table; + private byte[] cv = EMPTY_CV.getExpression(); + RyaTableMutationsFactory mut; + + public StatementToMutationMapper() { + } + + @Override + protected void setup(Context context) throws IOException, InterruptedException { + super.setup(context); + Configuration conf = context.getConfiguration(); + mut = new RyaTableMutationsFactory(RyaTripleContext.getInstance(new AccumuloRdfConfiguration(conf))); + tablePrefix = conf.get(MRUtils.TABLE_PREFIX_PROPERTY, RdfCloudTripleStoreConstants.TBL_PRFX_DEF); + spo_table = new Text(tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX); + po_table = new Text(tablePrefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX); + osp_table = new Text(tablePrefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX); + + final String cv_s = conf.get(MRUtils.AC_CV_PROP); + if (cv_s != null) + cv = cv_s.getBytes(); + } + + @Override + protected void map(LongWritable key, RyaStatementWritable value, Context context) throws IOException, InterruptedException { + RyaStatement statement = value.getRyaStatement(); + if (statement.getColumnVisibility() == null) { + statement.setColumnVisibility(cv); + } + Map<RdfCloudTripleStoreConstants.TABLE_LAYOUT, Collection<Mutation>> mutationMap = + mut.serialize(statement); + Collection<Mutation> spo = mutationMap.get(RdfCloudTripleStoreConstants.TABLE_LAYOUT.SPO); + Collection<Mutation> po = mutationMap.get(RdfCloudTripleStoreConstants.TABLE_LAYOUT.PO); + Collection<Mutation> osp = mutationMap.get(RdfCloudTripleStoreConstants.TABLE_LAYOUT.OSP); + + for (Mutation m : spo) { + context.write(spo_table, m); + } + for (Mutation m : po) { + context.write(po_table, m); + } + for (Mutation m : osp) { + context.write(osp_table, m); + } + } + + } +} + http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/upgrade/Upgrade322Tool.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/upgrade/Upgrade322Tool.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/upgrade/Upgrade322Tool.java new file mode 100644 index 0000000..68392b5 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/upgrade/Upgrade322Tool.java @@ -0,0 +1,239 @@ +package mvm.rya.accumulo.mr.upgrade; + +/* + * #%L + * mvm.rya.accumulo.rya + * %% + * Copyright (C) 2014 Rya + * %% + * 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. + * #L% + */ + +import mvm.rya.accumulo.mr.AbstractAccumuloMRTool; +import mvm.rya.accumulo.mr.utils.MRUtils; +import org.apache.accumulo.core.client.IteratorSetting; +import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Mutation; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.iterators.user.RegExFilter; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.Mapper; +import org.apache.hadoop.mapreduce.Reducer; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; +import org.calrissian.mango.types.LexiTypeEncoders; +import org.calrissian.mango.types.TypeEncoder; + +import java.io.IOException; +import java.util.Date; + +import static mvm.rya.api.RdfCloudTripleStoreConstants.*; + +/** + */ +public class Upgrade322Tool extends AbstractAccumuloMRTool implements Tool { + @Override + public int run(String[] strings) throws Exception { + conf.set(MRUtils.JOB_NAME_PROP, "Upgrade to Rya 3.2.2"); + //faster + init(); + + Job job = new Job(conf); + job.setJarByClass(Upgrade322Tool.class); + + setupInputFormat(job); + AccumuloInputFormat.setInputTableName(job, tablePrefix + TBL_OSP_SUFFIX); + + //we do not need to change any row that is a string, custom, or uri type + IteratorSetting regex = new IteratorSetting(30, "regex", + RegExFilter.class); + RegExFilter.setRegexs(regex, "\\w*" + TYPE_DELIM + "[\u0003|\u0008|\u0002]", null, null, null, false); + RegExFilter.setNegate(regex, true); + + // set input output of the particular job + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(Mutation.class); + + setupOutputFormat(job, tablePrefix + + TBL_SPO_SUFFIX); + + // set mapper and reducer classes + job.setMapperClass(Upgrade322Mapper.class); + job.setReducerClass(Reducer.class); + + // Submit the job + return job.waitForCompletion(true) ? 0 : 1; + } + + public static void main(String[] args) { + try { + ToolRunner.run(new Configuration(), new Upgrade322Tool(), args); + } catch (Exception e) { + e.printStackTrace(); + } + } + + /** + * Reading from the OSP table + */ + public static class Upgrade322Mapper extends Mapper<Key, Value, Text, Mutation> { + + private String tablePrefix; + private Text spoTable; + private Text poTable; + private Text ospTable; + + private final UpgradeObjectSerialization upgradeObjectSerialization; + + public Upgrade322Mapper() { + this(new UpgradeObjectSerialization()); + } + + public Upgrade322Mapper( + UpgradeObjectSerialization upgradeObjectSerialization) { + this.upgradeObjectSerialization = upgradeObjectSerialization; + } + + @Override + protected void setup( + Context context) throws IOException, InterruptedException { + super.setup(context); + + tablePrefix = context.getConfiguration().get( + MRUtils.TABLE_PREFIX_PROPERTY, TBL_PRFX_DEF); + spoTable = new Text(tablePrefix + TBL_SPO_SUFFIX); + poTable = new Text(tablePrefix + TBL_PO_SUFFIX); + ospTable = new Text(tablePrefix + TBL_OSP_SUFFIX); + } + + @Override + protected void map( + Key key, Value value, Context context) + throws IOException, InterruptedException { + + //read the key, expect OSP + final String row = key.getRow().toString(); + final int firstDelim = row.indexOf(DELIM); + final int secondDelim = row.indexOf(DELIM, firstDelim + 1); + final int typeDelim = row.lastIndexOf(TYPE_DELIM); + final String oldSerialization = row.substring(0, firstDelim); + char typeMarker = row.charAt(row.length() - 1); + + final String subject = row.substring(firstDelim + 1, secondDelim); + final String predicate = row.substring(secondDelim + 1, typeDelim); + final String typeSuffix = TYPE_DELIM + typeMarker; + + String newSerialization = upgradeObjectSerialization.upgrade(oldSerialization, typeMarker); + if(newSerialization == null) { + return; + } + + //write out delete Mutations + Mutation deleteOldSerialization_osp = new Mutation(key.getRow()); + deleteOldSerialization_osp.putDelete(key.getColumnFamily(), key.getColumnQualifier(), + key.getColumnVisibilityParsed()); + Mutation deleteOldSerialization_po = new Mutation(predicate + DELIM + oldSerialization + DELIM + subject + typeSuffix); + deleteOldSerialization_po.putDelete(key.getColumnFamily(), + key.getColumnQualifier(), + key.getColumnVisibilityParsed()); + Mutation deleteOldSerialization_spo = new Mutation(subject + DELIM + predicate + DELIM + oldSerialization + typeSuffix); + deleteOldSerialization_spo.putDelete(key.getColumnFamily(), key.getColumnQualifier(), + key.getColumnVisibilityParsed()); + + //write out new serialization + Mutation putNewSerialization_osp = new Mutation(newSerialization + DELIM + subject + DELIM + predicate + typeSuffix); + putNewSerialization_osp.put(key.getColumnFamily(), + key.getColumnQualifier(), + key.getColumnVisibilityParsed(), + key.getTimestamp(), value); + Mutation putNewSerialization_po = new Mutation(predicate + DELIM + newSerialization + DELIM + subject + typeSuffix); + putNewSerialization_po.put(key.getColumnFamily(), + key.getColumnQualifier(), + key.getColumnVisibilityParsed(), + key.getTimestamp(), value); + Mutation putNewSerialization_spo = new Mutation(subject + DELIM + predicate + DELIM + newSerialization + typeSuffix); + putNewSerialization_spo.put(key.getColumnFamily(), + key.getColumnQualifier(), + key.getColumnVisibilityParsed(), + key.getTimestamp(), value); + + //write out deletes to all tables + context.write(ospTable, deleteOldSerialization_osp); + context.write(poTable, deleteOldSerialization_po); + context.write(spoTable, deleteOldSerialization_spo); + + //write out inserts to all tables + context.write(ospTable, putNewSerialization_osp); + context.write(poTable, putNewSerialization_po); + context.write(spoTable, putNewSerialization_spo); + } + } + + public static class UpgradeObjectSerialization { + + public static final TypeEncoder<Boolean, String> + BOOLEAN_STRING_TYPE_ENCODER = LexiTypeEncoders.booleanEncoder(); + public static final TypeEncoder<Byte, String> BYTE_STRING_TYPE_ENCODER + = LexiTypeEncoders.byteEncoder(); + public static final TypeEncoder<Date, String> DATE_STRING_TYPE_ENCODER + = LexiTypeEncoders.dateEncoder(); + public static final TypeEncoder<Integer, String> + INTEGER_STRING_TYPE_ENCODER = LexiTypeEncoders.integerEncoder(); + public static final TypeEncoder<Long, String> LONG_STRING_TYPE_ENCODER + = LexiTypeEncoders.longEncoder(); + public static final TypeEncoder<Double, String> + DOUBLE_STRING_TYPE_ENCODER = LexiTypeEncoders.doubleEncoder(); + + public String upgrade(String object, int typeMarker) { + switch(typeMarker) { + case 10: //boolean + final boolean bool = Boolean.parseBoolean(object); + return BOOLEAN_STRING_TYPE_ENCODER.encode(bool); + case 9: //byte + final byte b = Byte.parseByte(object); + return BYTE_STRING_TYPE_ENCODER.encode(b); + case 4: //long + final Long lng = Long.parseLong(object); + return LONG_STRING_TYPE_ENCODER.encode(lng); + case 5: //int + final Integer i = Integer.parseInt(object); + return INTEGER_STRING_TYPE_ENCODER.encode(i); + case 6: //double + String exp = object.substring(2, 5); + char valueSign = object.charAt(0); + char expSign = object.charAt(1); + Integer expInt = Integer.parseInt(exp); + if (expSign == '-') { + expInt = 999 - expInt; + } + final String expDoubleStr = + String.format("%s%sE%s%d", valueSign, + object.substring(6), + expSign, expInt); + return DOUBLE_STRING_TYPE_ENCODER + .encode(Double.parseDouble(expDoubleStr)); + case 7: //datetime + //check to see if it is an early release that includes the exact term xsd:dateTime + final Long l = Long.MAX_VALUE - Long.parseLong(object); + Date date = new Date(l); + return DATE_STRING_TYPE_ENCODER.encode(date); + default: + return null; + } + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/AccumuloHDFSFileInputFormat.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/AccumuloHDFSFileInputFormat.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/AccumuloHDFSFileInputFormat.java new file mode 100644 index 0000000..399cf45 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/AccumuloHDFSFileInputFormat.java @@ -0,0 +1,205 @@ +package mvm.rya.accumulo.mr.utils; + +/* + * #%L + * mvm.rya.accumulo.rya + * %% + * Copyright (C) 2014 Rya + * %% + * 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. + * #L% + */ + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; + +import org.apache.accumulo.core.Constants; +import org.apache.accumulo.core.client.Connector; +import org.apache.accumulo.core.client.Instance; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.client.admin.TableOperations; +import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat; +import org.apache.accumulo.core.client.security.tokens.AuthenticationToken; +import org.apache.accumulo.core.client.security.tokens.PasswordToken; +import org.apache.accumulo.core.conf.AccumuloConfiguration; +import org.apache.accumulo.core.conf.Property; +import org.apache.accumulo.core.data.ByteSequence; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Range; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.file.FileSKVIterator; +import org.apache.accumulo.core.file.rfile.RFileOperations; +import org.apache.accumulo.core.util.ArgumentChecker; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.BlockLocation; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.JobContext; +import org.apache.hadoop.mapreduce.Mapper; +import org.apache.hadoop.mapreduce.RecordReader; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; +import org.apache.hadoop.mapreduce.lib.input.FileSplit; +import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat; + +/** + * Finds the accumulo tablet files on the hdfs disk, and uses that as the input for MR jobs + * Date: 5/11/12 + * Time: 2:04 PM + */ +public class AccumuloHDFSFileInputFormat extends FileInputFormat<Key, Value> { + + public static final Range ALLRANGE = new Range(new Text("\u0000"), new Text("\uFFFD")); + + @Override + public List<InputSplit> getSplits(JobContext jobContext) throws IOException { + //read the params from AccumuloInputFormat + Configuration conf = jobContext.getConfiguration(); + Instance instance = AccumuloProps.getInstance(jobContext); + String user = AccumuloProps.getUsername(jobContext); + AuthenticationToken password = AccumuloProps.getPassword(jobContext); + String table = AccumuloProps.getTablename(jobContext); + ArgumentChecker.notNull(instance); + ArgumentChecker.notNull(table); + + //find the files necessary + try { + AccumuloConfiguration acconf = instance.getConfiguration(); + FileSystem fs = FileSystem.get(conf); + Connector connector = instance.getConnector(user, password); + TableOperations tos = connector.tableOperations(); + String tableId = tos.tableIdMap().get(table); + String filePrefix = acconf.get(Property.INSTANCE_DFS_DIR) + "/tables/" + tableId; + System.out.println(filePrefix); + + Scanner scanner = connector.createScanner("!METADATA", Constants.NO_AUTHS); //TODO: auths? + scanner.setRange(new Range(new Text(tableId + "\u0000"), new Text(tableId + "\uFFFD"))); + scanner.fetchColumnFamily(new Text("file")); + List<String> files = new ArrayList<String>(); + List<InputSplit> fileSplits = new ArrayList<InputSplit>(); + Job job = new Job(conf); + for (Map.Entry<Key, Value> entry : scanner) { + String file = filePrefix + entry.getKey().getColumnQualifier().toString(); + files.add(file); + Path path = new Path(file); + FileStatus fileStatus = fs.getFileStatus(path); + long len = fileStatus.getLen(); + BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(fileStatus, 0, len); + fileSplits.add(new FileSplit(path, 0, len, fileBlockLocations[0].getHosts())); +// FileInputFormat.addInputPath(job, path); + } + System.out.println(files); + return fileSplits; +// return super.getSplits(job); + } catch (Exception e) { + throw new IOException(e); + } + } + + @Override + public RecordReader<Key, Value> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { + return new RecordReader<Key, Value>() { + + private FileSKVIterator fileSKVIterator; + + @Override + public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { + FileSplit split = (FileSplit) inputSplit; + Configuration job = taskAttemptContext.getConfiguration(); + Path file = split.getPath(); +// long start = split.getStart(); +// long length = split.getLength(); + FileSystem fs = file.getFileSystem(job); +// FSDataInputStream fileIn = fs.open(file); +// System.out.println(start); +// if (start != 0L) { +// fileIn.seek(start); +// } + Instance instance = AccumuloProps.getInstance(taskAttemptContext); + + fileSKVIterator = RFileOperations.getInstance().openReader(file.toString(), ALLRANGE, + new HashSet<ByteSequence>(), false, fs, job, instance.getConfiguration()); +// fileSKVIterator = new RFileOperations2().openReader(fileIn, length - start, job); + } + + @Override + public boolean nextKeyValue() throws IOException, InterruptedException { + fileSKVIterator.next(); + return fileSKVIterator.hasTop(); + } + + @Override + public Key getCurrentKey() throws IOException, InterruptedException { + return fileSKVIterator.getTopKey(); + } + + @Override + public Value getCurrentValue() throws IOException, InterruptedException { + return fileSKVIterator.getTopValue(); + } + + @Override + public float getProgress() throws IOException, InterruptedException { + return 0; + } + + @Override + public void close() throws IOException { + //To change body of implemented methods use File | Settings | File Templates. + } + }; + } + + public static void main(String[] args) { + try { + Job job = new Job(new Configuration()); + job.setJarByClass(AccumuloHDFSFileInputFormat.class); + Configuration conf = job.getConfiguration(); + conf.setBoolean("mapred.map.tasks.speculative.execution", false); + conf.setBoolean("mapred.reduce.tasks.speculative.execution", false); + AccumuloInputFormat.setConnectorInfo(job, "root", new PasswordToken("secret")); + AccumuloInputFormat.setInputTableName(job, "l_spo"); + AccumuloInputFormat.setScanAuthorizations(job, Constants.NO_AUTHS); + AccumuloInputFormat.setZooKeeperInstance(job, "acu13", "stratus25:2181"); + AccumuloInputFormat.setRanges(job, Collections.singleton(ALLRANGE)); + job.setMapperClass(NullMapper.class); + job.setNumReduceTasks(0); + job.setOutputFormatClass(NullOutputFormat.class); + if (args.length == 0) { + job.setInputFormatClass(AccumuloHDFSFileInputFormat.class); + } else { + job.setInputFormatClass(AccumuloInputFormat.class); + } + job.waitForCompletion(true); + } catch (Exception e) { + e.printStackTrace(); + } + } + + @SuppressWarnings("rawtypes") + public static class NullMapper extends Mapper { + @Override + protected void map(Object key, Object value, Context context) throws IOException, InterruptedException { + + } + } +} + http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/AccumuloProps.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/AccumuloProps.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/AccumuloProps.java new file mode 100644 index 0000000..f98bf7f --- /dev/null +++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/AccumuloProps.java @@ -0,0 +1,57 @@ +package mvm.rya.accumulo.mr.utils; + +/* + * #%L + * mvm.rya.accumulo.rya + * %% + * Copyright (C) 2014 Rya + * %% + * 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. + * #L% + */ + +import java.io.IOException; + +import org.apache.accumulo.core.client.Instance; +import org.apache.accumulo.core.client.mapreduce.InputFormatBase; +import org.apache.accumulo.core.client.security.tokens.AuthenticationToken; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.hadoop.mapreduce.JobContext; +import org.apache.hadoop.mapreduce.RecordReader; +import org.apache.hadoop.mapreduce.TaskAttemptContext; + +@SuppressWarnings("rawtypes") +public class AccumuloProps extends InputFormatBase { + + @Override + public RecordReader createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { + throw new UnsupportedOperationException("Accumulo Props just holds properties"); + } + + public static Instance getInstance(JobContext conf) { + return InputFormatBase.getInstance(conf); + } + + public static AuthenticationToken getPassword(JobContext conf) { + return InputFormatBase.getAuthenticationToken(conf); + } + + public static String getUsername(JobContext conf) { + return InputFormatBase.getPrincipal(conf); + } + + public static String getTablename(JobContext conf) { + return InputFormatBase.getInputTableName(conf); + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/MRUtils.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/MRUtils.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/MRUtils.java new file mode 100644 index 0000000..6d53172 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/mr/utils/MRUtils.java @@ -0,0 +1,118 @@ +package mvm.rya.accumulo.mr.utils; + +/* + * #%L + * mvm.rya.accumulo.rya + * %% + * Copyright (C) 2014 Rya + * %% + * 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. + * #L% + */ + +import org.apache.hadoop.conf.Configuration; +import org.openrdf.model.URI; +import org.openrdf.model.ValueFactory; +import org.openrdf.model.impl.ValueFactoryImpl; + +/** + * Class MRSailUtils + * Date: May 19, 2011 + * Time: 10:34:06 AM + */ +public class MRUtils { + + public static final String JOB_NAME_PROP = "mapred.job.name"; + + public static final String AC_USERNAME_PROP = "ac.username"; + public static final String AC_PWD_PROP = "ac.pwd"; + public static final String AC_ZK_PROP = "ac.zk"; + public static final String AC_INSTANCE_PROP = "ac.instance"; + public static final String AC_TTL_PROP = "ac.ttl"; + public static final String AC_TABLE_PROP = "ac.table"; + public static final String AC_AUTH_PROP = "ac.auth"; + public static final String AC_CV_PROP = "ac.cv"; + public static final String AC_MOCK_PROP = "ac.mock"; + public static final String AC_HDFS_INPUT_PROP = "ac.hdfsinput"; + public static final String HADOOP_IO_SORT_MB = "ac.hdfsinput"; + public static final String TABLE_LAYOUT_PROP = "rdf.tablelayout"; + public static final String FORMAT_PROP = "rdf.format"; + public static final String INPUT_PATH = "input"; + + public static final String NAMED_GRAPH_PROP = "rdf.graph"; + + public static final String TABLE_PREFIX_PROPERTY = "rdf.tablePrefix"; + + // rdf constants + public static final ValueFactory vf = new ValueFactoryImpl(); + public static final URI RDF_TYPE = vf.createURI("http://www.w3.org/1999/02/22-rdf-syntax-ns#";, "type"); + + + // cloudbase map reduce utils + +// public static Range retrieveRange(URI entry_key, URI entry_val) throws IOException { +// ByteArrayDataOutput startRowOut = ByteStreams.newDataOutput(); +// startRowOut.write(RdfCloudTripleStoreUtils.writeValue(entry_key)); +// if (entry_val != null) { +// startRowOut.write(RdfCloudTripleStoreConstants.DELIM_BYTES); +// startRowOut.write(RdfCloudTripleStoreUtils.writeValue(entry_val)); +// } +// byte[] startrow = startRowOut.toByteArray(); +// startRowOut.write(RdfCloudTripleStoreConstants.DELIM_STOP_BYTES); +// byte[] stoprow = startRowOut.toByteArray(); +// +// Range range = new Range(new Text(startrow), new Text(stoprow)); +// return range; +// } + + + public static String getACTtl(Configuration conf) { + return conf.get(AC_TTL_PROP); + } + + public static String getACUserName(Configuration conf) { + return conf.get(AC_USERNAME_PROP); + } + + public static String getACPwd(Configuration conf) { + return conf.get(AC_PWD_PROP); + } + + public static String getACZK(Configuration conf) { + return conf.get(AC_ZK_PROP); + } + + public static String getACInstance(Configuration conf) { + return conf.get(AC_INSTANCE_PROP); + } + + public static void setACUserName(Configuration conf, String str) { + conf.set(AC_USERNAME_PROP, str); + } + + public static void setACPwd(Configuration conf, String str) { + conf.set(AC_PWD_PROP, str); + } + + public static void setACZK(Configuration conf, String str) { + conf.set(AC_ZK_PROP, str); + } + + public static void setACInstance(Configuration conf, String str) { + conf.set(AC_INSTANCE_PROP, str); + } + + public static void setACTtl(Configuration conf, String str) { + conf.set(AC_TTL_PROP, str); + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/AccumuloRyaQueryEngine.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/AccumuloRyaQueryEngine.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/AccumuloRyaQueryEngine.java new file mode 100644 index 0000000..2b6f182 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/AccumuloRyaQueryEngine.java @@ -0,0 +1,400 @@ +package mvm.rya.accumulo.query; + +/* + * #%L + * mvm.rya.accumulo.rya + * %% + * Copyright (C) 2014 Rya + * %% + * 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. + * #L% + */ + +import static mvm.rya.api.RdfCloudTripleStoreUtils.layoutToTable; +import info.aduna.iteration.CloseableIteration; + +import java.io.IOException; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.Map; + +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.api.RdfCloudTripleStoreConfiguration; +import mvm.rya.api.RdfCloudTripleStoreConstants; +import mvm.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT; +import mvm.rya.api.RdfCloudTripleStoreUtils; +import mvm.rya.api.domain.RyaRange; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.domain.RyaType; +import mvm.rya.api.domain.RyaURI; +import mvm.rya.api.layout.TableLayoutStrategy; +import mvm.rya.api.persist.RyaDAOException; +import mvm.rya.api.persist.query.BatchRyaQuery; +import mvm.rya.api.persist.query.RyaQuery; +import mvm.rya.api.persist.query.RyaQueryEngine; +import mvm.rya.api.query.strategy.ByteRange; +import mvm.rya.api.query.strategy.TriplePatternStrategy; +import mvm.rya.api.resolver.RyaContext; +import mvm.rya.api.resolver.RyaTripleContext; +import mvm.rya.api.resolver.triple.TripleRowRegex; +import mvm.rya.api.utils.CloseableIterableIteration; +import mvm.rya.iterators.LimitingAgeOffFilter; + +import org.apache.accumulo.core.client.BatchScanner; +import org.apache.accumulo.core.client.Connector; +import org.apache.accumulo.core.client.IteratorSetting; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.client.ScannerBase; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Range; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.iterators.user.RegExFilter; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.hadoop.io.Text; +import org.calrissian.mango.collect.CloseableIterable; +import org.calrissian.mango.collect.CloseableIterables; +import org.calrissian.mango.collect.FluentCloseableIterable; +import org.openrdf.query.BindingSet; + +import com.google.common.base.Function; +import com.google.common.base.Preconditions; +import com.google.common.collect.FluentIterable; +import com.google.common.collect.Iterators; + +/** + * Date: 7/17/12 + * Time: 9:28 AM + */ +public class AccumuloRyaQueryEngine implements RyaQueryEngine<AccumuloRdfConfiguration> { + + private AccumuloRdfConfiguration configuration; + private Connector connector; + private RyaTripleContext ryaContext; + private final Map<TABLE_LAYOUT, KeyValueToRyaStatementFunction> keyValueToRyaStatementFunctionMap = new HashMap<TABLE_LAYOUT, KeyValueToRyaStatementFunction>(); + + public AccumuloRyaQueryEngine(Connector connector) { + this(connector, new AccumuloRdfConfiguration()); + } + + public AccumuloRyaQueryEngine(Connector connector, AccumuloRdfConfiguration conf) { + this.connector = connector; + this.configuration = conf; + ryaContext = RyaTripleContext.getInstance(conf); + keyValueToRyaStatementFunctionMap.put(TABLE_LAYOUT.SPO, new KeyValueToRyaStatementFunction(TABLE_LAYOUT.SPO, ryaContext)); + keyValueToRyaStatementFunctionMap.put(TABLE_LAYOUT.PO, new KeyValueToRyaStatementFunction(TABLE_LAYOUT.PO, ryaContext)); + keyValueToRyaStatementFunctionMap.put(TABLE_LAYOUT.OSP, new KeyValueToRyaStatementFunction(TABLE_LAYOUT.OSP, ryaContext)); + } + + @Override + public CloseableIteration<RyaStatement, RyaDAOException> query(RyaStatement stmt, AccumuloRdfConfiguration conf) throws RyaDAOException { + if (conf == null) { + conf = configuration; + } + + RyaQuery ryaQuery = RyaQuery.builder(stmt).load(conf).build(); + CloseableIterable<RyaStatement> results = query(ryaQuery); + + return new CloseableIterableIteration<RyaStatement, RyaDAOException>(results); + } + + protected String getData(RyaType ryaType) { + return (ryaType != null) ? (ryaType.getData()) : (null); + } + + @Override + public CloseableIteration<? extends Map.Entry<RyaStatement, BindingSet>, RyaDAOException> queryWithBindingSet(Collection<Map.Entry<RyaStatement, BindingSet>> stmts, AccumuloRdfConfiguration conf) throws RyaDAOException { + if (conf == null) { + conf = configuration; + } + //query configuration + Authorizations authorizations = conf.getAuthorizations(); + Long ttl = conf.getTtl(); + Long maxResults = conf.getLimit(); + Integer maxRanges = conf.getMaxRangesForScanner(); + Integer numThreads = conf.getNumThreads(); + + //TODO: cannot span multiple tables here + try { + Collection<Range> ranges = new HashSet<Range>(); + RangeBindingSetEntries rangeMap = new RangeBindingSetEntries(); + TABLE_LAYOUT layout = null; + RyaURI context = null; + TriplePatternStrategy strategy = null; + for (Map.Entry<RyaStatement, BindingSet> stmtbs : stmts) { + RyaStatement stmt = stmtbs.getKey(); + context = stmt.getContext(); //TODO: This will be overwritten + BindingSet bs = stmtbs.getValue(); + strategy = ryaContext.retrieveStrategy(stmt); + if (strategy == null) { + throw new IllegalArgumentException("TriplePattern[" + stmt + "] not supported"); + } + + Map.Entry<RdfCloudTripleStoreConstants.TABLE_LAYOUT, ByteRange> entry = + strategy.defineRange(stmt.getSubject(), stmt.getPredicate(), stmt.getObject(), stmt.getContext(), conf); + + //use range to set scanner + //populate scanner based on authorizations, ttl + layout = entry.getKey(); + ByteRange byteRange = entry.getValue(); + Range range = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); + ranges.add(range); + rangeMap.ranges.add(new RdfCloudTripleStoreUtils.CustomEntry<Range, BindingSet>(range, bs)); + } + //no ranges + if (layout == null) return null; + String regexSubject = conf.getRegexSubject(); + String regexPredicate = conf.getRegexPredicate(); + String regexObject = conf.getRegexObject(); + TripleRowRegex tripleRowRegex = strategy.buildRegex(regexSubject, regexPredicate, regexObject, null, null); + + String table = layoutToTable(layout, conf); + boolean useBatchScanner = ranges.size() > maxRanges; + RyaStatementBindingSetKeyValueIterator iterator = null; + if (useBatchScanner) { + ScannerBase scanner = connector.createBatchScanner(table, authorizations, numThreads); + ((BatchScanner) scanner).setRanges(ranges); + fillScanner(scanner, context, null, ttl, null, tripleRowRegex, conf); + iterator = new RyaStatementBindingSetKeyValueIterator(layout, ryaContext, scanner, rangeMap); + } else { + Scanner scannerBase = null; + Iterator<Map.Entry<Key, Value>>[] iters = new Iterator[ranges.size()]; + int i = 0; + for (Range range : ranges) { + scannerBase = connector.createScanner(table, authorizations); + scannerBase.setRange(range); + fillScanner(scannerBase, context, null, ttl, null, tripleRowRegex, conf); + iters[i] = scannerBase.iterator(); + i++; + } + iterator = new RyaStatementBindingSetKeyValueIterator(layout, Iterators.concat(iters), rangeMap, ryaContext); + } + if (maxResults != null) { + iterator.setMaxResults(maxResults); + } + return iterator; + } catch (Exception e) { + throw new RyaDAOException(e); + } + + } + + @Override + public CloseableIteration<RyaStatement, RyaDAOException> batchQuery(Collection<RyaStatement> stmts, AccumuloRdfConfiguration conf) + throws RyaDAOException { + if (conf == null) { + conf = configuration; + } + + BatchRyaQuery batchRyaQuery = BatchRyaQuery.builder(stmts).load(conf).build(); + CloseableIterable<RyaStatement> results = query(batchRyaQuery); + + return new CloseableIterableIteration<RyaStatement, RyaDAOException>(results); + } + + @Override + public CloseableIterable<RyaStatement> query(RyaQuery ryaQuery) throws RyaDAOException { + Preconditions.checkNotNull(ryaQuery); + RyaStatement stmt = ryaQuery.getQuery(); + Preconditions.checkNotNull(stmt); + + //query configuration + String[] auths = ryaQuery.getAuths(); + Authorizations authorizations = auths != null ? new Authorizations(auths) : configuration.getAuthorizations(); + Long ttl = ryaQuery.getTtl(); + Long currentTime = ryaQuery.getCurrentTime(); + Long maxResults = ryaQuery.getMaxResults(); + Integer batchSize = ryaQuery.getBatchSize(); + String regexSubject = ryaQuery.getRegexSubject(); + String regexPredicate = ryaQuery.getRegexPredicate(); + String regexObject = ryaQuery.getRegexObject(); + TableLayoutStrategy tableLayoutStrategy = configuration.getTableLayoutStrategy(); + + try { + //find triple pattern range + TriplePatternStrategy strategy = ryaContext.retrieveStrategy(stmt); + TABLE_LAYOUT layout; + Range range; + RyaURI subject = stmt.getSubject(); + RyaURI predicate = stmt.getPredicate(); + RyaType object = stmt.getObject(); + RyaURI context = stmt.getContext(); + String qualifier = stmt.getQualifer(); + TripleRowRegex tripleRowRegex = null; + if (strategy != null) { + //otherwise, full table scan is supported + Map.Entry<RdfCloudTripleStoreConstants.TABLE_LAYOUT, ByteRange> entry = + strategy.defineRange(subject, predicate, object, context, null); + layout = entry.getKey(); + ByteRange byteRange = entry.getValue(); + range = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); + + byte[] objectTypeInfo = null; + if (object != null) { + //TODO: Not good to serialize this twice + if (object instanceof RyaRange) { + objectTypeInfo = RyaContext.getInstance().serializeType(((RyaRange) object).getStart())[1]; + } else { + objectTypeInfo = RyaContext.getInstance().serializeType(object)[1]; + } + } + + tripleRowRegex = strategy.buildRegex(regexSubject, regexPredicate, regexObject, null, objectTypeInfo); + } else { + range = new Range(); + layout = TABLE_LAYOUT.SPO; + } + + //use range to set scanner + //populate scanner based on authorizations, ttl + String table = layoutToTable(layout, tableLayoutStrategy); + Scanner scanner = connector.createScanner(table, authorizations); + scanner.setRange(range); + if (batchSize != null) { + scanner.setBatchSize(batchSize); + } + fillScanner(scanner, context, qualifier, ttl, currentTime, tripleRowRegex, ryaQuery.getConf()); + + FluentCloseableIterable<RyaStatement> results = FluentCloseableIterable.from(new ScannerBaseCloseableIterable(scanner)) + .transform(keyValueToRyaStatementFunctionMap.get(layout)); + if (maxResults != null) { + results = results.limit(maxResults.intValue()); + } + + return results; + } catch (Exception e) { + throw new RyaDAOException(e); + } + } + + @Override + public CloseableIterable<RyaStatement> query(BatchRyaQuery ryaQuery) throws RyaDAOException { + Preconditions.checkNotNull(ryaQuery); + Iterable<RyaStatement> stmts = ryaQuery.getQueries(); + Preconditions.checkNotNull(stmts); + + //query configuration + String[] auths = ryaQuery.getAuths(); + final Authorizations authorizations = auths != null ? new Authorizations(auths) : configuration.getAuthorizations(); + final Long ttl = ryaQuery.getTtl(); + Long currentTime = ryaQuery.getCurrentTime(); + Long maxResults = ryaQuery.getMaxResults(); + Integer batchSize = ryaQuery.getBatchSize(); + Integer numQueryThreads = ryaQuery.getNumQueryThreads(); + String regexSubject = ryaQuery.getRegexSubject(); + String regexPredicate = ryaQuery.getRegexPredicate(); + String regexObject = ryaQuery.getRegexObject(); + TableLayoutStrategy tableLayoutStrategy = configuration.getTableLayoutStrategy(); + int maxRanges = ryaQuery.getMaxRanges(); + + //TODO: cannot span multiple tables here + try { + Collection<Range> ranges = new HashSet<Range>(); + TABLE_LAYOUT layout = null; + RyaURI context = null; + TriplePatternStrategy strategy = null; + for (RyaStatement stmt : stmts) { + context = stmt.getContext(); //TODO: This will be overwritten + strategy = ryaContext.retrieveStrategy(stmt); + if (strategy == null) { + throw new IllegalArgumentException("TriplePattern[" + stmt + "] not supported"); + } + + Map.Entry<RdfCloudTripleStoreConstants.TABLE_LAYOUT, ByteRange> entry = + strategy.defineRange(stmt.getSubject(), stmt.getPredicate(), stmt.getObject(), stmt.getContext(), null); + + //use range to set scanner + //populate scanner based on authorizations, ttl + layout = entry.getKey(); + ByteRange byteRange = entry.getValue(); + Range range = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); + ranges.add(range); + } + //no ranges + if (layout == null) throw new IllegalArgumentException("No table layout specified"); + + final TripleRowRegex tripleRowRegex = strategy.buildRegex(regexSubject, regexPredicate, regexObject, null, null); + + final String table = layoutToTable(layout, tableLayoutStrategy); + boolean useBatchScanner = ranges.size() > maxRanges; + FluentCloseableIterable<RyaStatement> results = null; + if (useBatchScanner) { + BatchScanner scanner = connector.createBatchScanner(table, authorizations, numQueryThreads); + scanner.setRanges(ranges); + fillScanner(scanner, context, null, ttl, null, tripleRowRegex, ryaQuery.getConf()); + results = FluentCloseableIterable.from(new ScannerBaseCloseableIterable(scanner)).transform(keyValueToRyaStatementFunctionMap.get(layout)); + } else { + final RyaURI fcontext = context; + final RdfCloudTripleStoreConfiguration fconf = ryaQuery.getConf(); + FluentIterable<RyaStatement> fluent = FluentIterable.from(ranges).transformAndConcat(new Function<Range, Iterable<Map.Entry<Key, Value>>>() { + @Override + public Iterable<Map.Entry<Key, Value>> apply(Range range) { + try { + Scanner scanner = connector.createScanner(table, authorizations); + scanner.setRange(range); + fillScanner(scanner, fcontext, null, ttl, null, tripleRowRegex, fconf); + return scanner; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + }).transform(keyValueToRyaStatementFunctionMap.get(layout)); + + results = FluentCloseableIterable.from(CloseableIterables.wrap(fluent)); + } + if (maxResults != null) { + results = results.limit(maxResults.intValue()); + } + return results; + } catch (Exception e) { + throw new RyaDAOException(e); + } + } + + protected void fillScanner(ScannerBase scanner, RyaURI context, String qualifier, Long ttl, Long currentTime, TripleRowRegex tripleRowRegex, RdfCloudTripleStoreConfiguration conf) throws IOException { + if (context != null && qualifier != null) { + scanner.fetchColumn(new Text(context.getData()), new Text(qualifier)); + } else if (context != null) { + scanner.fetchColumnFamily(new Text(context.getData())); + } else if (qualifier != null) { + IteratorSetting setting = new IteratorSetting(8, "riq", RegExFilter.class.getName()); + RegExFilter.setRegexs(setting, null, null, qualifier, null, false); + scanner.addScanIterator(setting); + } + if (ttl != null) { + IteratorSetting setting = new IteratorSetting(9, "fi", LimitingAgeOffFilter.class.getName()); + setting.addOption(LimitingAgeOffFilter.TTL, Long.toString(ttl)); + if(currentTime != null){ + setting.addOption(LimitingAgeOffFilter.CURRENT_TIME, Long.toString(currentTime)); + } + scanner.addScanIterator(setting); + } + if (tripleRowRegex != null) { + IteratorSetting setting = new IteratorSetting(11, "ri", RegExFilter.class.getName()); + String regex = tripleRowRegex.getRow(); + RegExFilter.setRegexs(setting, regex, null, null, null, false); + scanner.addScanIterator(setting); + } + } + + @Override + public void setConf(AccumuloRdfConfiguration conf) { + this.configuration = conf; + } + + @Override + public AccumuloRdfConfiguration getConf() { + return configuration; + } +}
