You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rya.apache.org by ca...@apache.org on 2017/04/13 13:33:13 UTC
[2/3] incubator-rya git commit: RYA-253 Converted the rya.prospector
class from Groovy to Java. Also added documentation to the project. Closes
#151
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/domain/IntermediateProspect.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/domain/IntermediateProspect.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/domain/IntermediateProspect.java
new file mode 100644
index 0000000..8c52378
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/domain/IntermediateProspect.java
@@ -0,0 +1,213 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.domain;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.rya.prospector.mr.Prospector;
+import org.apache.rya.prospector.plans.IndexWorkPlan;
+
+/**
+ * Represents a piece of information that is being counted during the process
+ * of running a {@link Prospector} job.
+ */
+public class IntermediateProspect implements WritableComparable<IntermediateProspect> {
+
+ private String index;
+ private String data;
+ private String dataType;
+ private TripleValueType tripleValueType;
+ private String visibility;
+
+ /**
+ * Constructs an uninitialized instance of {@link IntermediateProspect}.
+ * This constructor is required to integration with Map Reduce's
+ * {@link WritableComparable} interface.
+ */
+ public IntermediateProspect() { }
+
+ /**
+ * Constructs an instance of {@link IntermediateProspect}.
+ *
+ * @param index - Indicates which {@link IndexWorkPlan} the data is part of.
+ * @param data - The information that is being counted.
+ * @param dataType - The data type of {@code data}.
+ * @param tripleValueType - Indicates which parts of the RDF Statement are included in {@code data}.
+ * @param visibility - The visibility of this entry.
+ */
+ public IntermediateProspect(
+ final String index,
+ final String data,
+ final String dataType,
+ final TripleValueType tripleValueType,
+ final String visibility) {
+ this.index = index;
+ this.data = data;
+ this.dataType = dataType;
+ this.tripleValueType = tripleValueType;
+ this.visibility = visibility;
+ }
+
+ /**
+ * @return Indicates which {@link IndexWorkPlan} the data is part of.
+ */
+ public String getIndex() {
+ return index;
+ }
+
+ /**
+ * @return The information that is being counted.
+ */
+ public String getData() {
+ return data;
+ }
+
+ /**
+ * @return The data type of {@code data}.
+ */
+ public String getDataType() {
+ return dataType;
+ }
+
+ /**
+ * @return Indicates which parts of the RDF Statement are included in {@code data}.
+ */
+ public TripleValueType getTripleValueType() {
+ return tripleValueType;
+ }
+
+ /**
+ * @return The visibility of this entry.
+ */
+ public String getVisibility() {
+ return visibility;
+ }
+
+ @Override
+ public int compareTo(IntermediateProspect t) {
+ if(!index.equals(t.index)) {
+ return index.compareTo(t.index);
+ }
+ if(!data.equals(t.data)) {
+ return data.compareTo(t.data);
+ }
+ if(!dataType.equals(t.dataType)) {
+ return dataType.compareTo(t.dataType);
+ }
+ if(!tripleValueType.equals(t.tripleValueType)) {
+ return tripleValueType.compareTo(t.tripleValueType);
+ }
+ if(!visibility.equals(t.visibility)) {
+ return visibility.compareTo(t.visibility);
+ }
+ return 0;
+ }
+
+ @Override
+ public void write(DataOutput dataOutput) throws IOException {
+ dataOutput.writeUTF(index);
+ dataOutput.writeUTF(data);
+ dataOutput.writeUTF(dataType);
+ dataOutput.writeUTF(tripleValueType.name());
+ dataOutput.writeUTF(visibility);
+ }
+
+ @Override
+ public void readFields(DataInput dataInput) throws IOException {
+ index = dataInput.readUTF();
+ data = dataInput.readUTF();
+ dataType = dataInput.readUTF();
+ tripleValueType = TripleValueType.valueOf(dataInput.readUTF());
+ visibility = dataInput.readUTF();
+ }
+
+ /**
+ * @return An empty instance of {@link Builder}.
+ */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /**
+ * Builds instances of {@link IntermediateProspect}.
+ */
+ public static final class Builder {
+
+ private String index;
+ private String data;
+ private String dataType;
+ private TripleValueType tripleValueType;
+ private String visibility;
+
+ /**
+ * @param index - Indicates which {@link IndexWorkPlan} the data is part of.
+ * @return This {@link Builder} so that method invocations may be chained.
+ */
+ public Builder setIndex(String index) {
+ this.index = index;
+ return this;
+ }
+
+ /**
+ * @param data - The information that is being counted.
+ * @return This {@link Builder} so that method invocations may be chained.
+ */
+ public Builder setData(String data) {
+ this.data = data;
+ return this;
+ }
+
+ /**
+ * @param dataType - The data type of {@code data}.
+ * @return This {@link Builder} so that method invocations may be chained.
+ */
+ public Builder setDataType(String dataType) {
+ this.dataType = dataType;
+ return this;
+ }
+
+ /**
+ * @param tripleValueType - Indicates which parts of the RDF Statement are included in {@code data}.
+ * @return This {@link Builder} so that method invocations may be chained.
+ */
+ public Builder setTripleValueType(TripleValueType tripleValueType) {
+ this.tripleValueType = tripleValueType;
+ return this;
+ }
+
+ /**
+ * @param visibility - The visibility of this entry.
+ * @return This {@link Builder} so that method invocations may be chained.
+ */
+ public Builder setVisibility(String visibility) {
+ this.visibility = visibility;
+ return this;
+ }
+
+ /**
+ * @return Constructs an instance of {@link IntermediateProspect} built using this builder's values.
+ */
+ public IntermediateProspect build() {
+ return new IntermediateProspect(index, data, dataType, tripleValueType, visibility);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/domain/TripleValueType.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/domain/TripleValueType.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/domain/TripleValueType.java
new file mode 100644
index 0000000..16e7916
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/domain/TripleValueType.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.domain;
+
+import static java.util.Objects.requireNonNull;
+
+import com.google.common.collect.ImmutableMap;
+
+/**
+ * Enumerates the different types of counts that are performed over a Rya instance's
+ * Statements as part of a Prospector run.
+ */
+public enum TripleValueType {
+ /**
+ * The data portion of an {@link IndexEntry} contains a unique Subject that
+ * appears within a Rya instance's Statements.
+ */
+ SUBJECT("subject"),
+
+ /**
+ * The data portion of an {@link IndexEntry} contains a unique Predicate that
+ * appears within a Rya instance's Statements.
+ */
+ PREDICATE("predicate"),
+
+ /**
+ * The data portion of an {@link IndexEntry} contains a unique Object that
+ * appears within a Rya instance's Statements.
+ */
+ OBJECT("object"),
+
+ /**
+ * The data portion of an {@link IndexEntrY} contains a unique Namespace from
+ * the Subjects that appear within a Rya instance.
+ */
+ ENTITY("entity"),
+
+ /**
+ * The data portion of an {@link IndexEntry} contains a unique Subject and Predicate
+ * pair that appears within a Rya instance's Statements.
+ */
+ SUBJECT_PREDICATE("subjectpredicate"),
+
+ /**
+ * The data portion of an {@link IndexEntry} contains a unique Predicate and Object
+ * pair that appears within a Rya instance's Statements.
+ */
+ PREDICATE_OBJECT("predicateobject"),
+
+ /**
+ * The data portion of an {@link IndexEntry} contains a unique Subject and Object
+ * pair that appears within a Rya instance's Statements.
+ */
+ SUBJECT_OBJECT("subjectobject");
+
+ private final String indexType;
+
+ private TripleValueType(String indexType) {
+ this.indexType = requireNonNull(indexType);
+ }
+
+ /**
+ * @return The Prospector Index Type represented by the enum value.
+ */
+ public String getIndexType() {
+ return indexType;
+ }
+
+ private static final ImmutableMap<String, TripleValueType> lookup;
+ static {
+ ImmutableMap.Builder<String, TripleValueType> builder = ImmutableMap.builder();
+ for(TripleValueType type : TripleValueType.values()) {
+ builder.put(type.getIndexType(), type);
+ }
+ lookup = builder.build();
+ }
+
+ /**
+ * @param indexType - The index name to lookup.
+ * @return The enum value that represents the index name.
+ */
+ public TripleValueType fromIndexType(String indexType) {
+ return lookup.get(indexType);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/Prospector.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/Prospector.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/Prospector.java
new file mode 100644
index 0000000..78ea371
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/Prospector.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.mr;
+
+import static org.apache.rya.prospector.utils.ProspectorConstants.DEFAULT_VIS;
+import static org.apache.rya.prospector.utils.ProspectorConstants.EMPTY;
+import static org.apache.rya.prospector.utils.ProspectorConstants.METADATA;
+import static org.apache.rya.prospector.utils.ProspectorConstants.PERFORMANT;
+import static org.apache.rya.prospector.utils.ProspectorConstants.PROSPECT_TIME;
+import static org.apache.rya.prospector.utils.ProspectorUtils.connector;
+import static org.apache.rya.prospector.utils.ProspectorUtils.getReverseIndexDateTime;
+import static org.apache.rya.prospector.utils.ProspectorUtils.instance;
+import static org.apache.rya.prospector.utils.ProspectorUtils.writeMutations;
+
+import java.util.Calendar;
+import java.util.Collections;
+import java.util.Date;
+
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.commons.lang.time.DateUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.rya.prospector.domain.IntermediateProspect;
+import org.apache.rya.prospector.utils.ProspectorUtils;
+
+/**
+ * Configures and runs the Hadoop Map Reduce job that executes the Prospector's work.
+ */
+public class Prospector extends Configured implements Tool {
+
+ private static long NOW = System.currentTimeMillis();
+
+ private Date truncatedDate;
+
+ public static void main(String[] args) throws Exception {
+ final int res = ToolRunner.run(new Prospector(), args);
+ System.exit(res);
+ }
+
+ @Override
+ public int run(String[] args) throws Exception {
+ final Configuration conf = getConf();
+
+ truncatedDate = DateUtils.truncate(new Date(NOW), Calendar.MINUTE);
+
+ final Path configurationPath = new Path(args[0]);
+ conf.addResource(configurationPath);
+
+ final String inTable = conf.get("prospector.intable");
+ final String outTable = conf.get("prospector.outtable");
+ final String auths_str = conf.get("prospector.auths");
+ assert inTable != null;
+ assert outTable != null;
+ assert auths_str != null;
+
+ final Job job = new Job(getConf(), this.getClass().getSimpleName() + "_" + System.currentTimeMillis());
+ job.setJarByClass(this.getClass());
+
+ final String[] auths = auths_str.split(",");
+ ProspectorUtils.initMRJob(job, inTable, outTable, auths);
+
+ job.getConfiguration().setLong("DATE", NOW);
+
+ final String performant = conf.get(PERFORMANT);
+ if (Boolean.parseBoolean(performant)) {
+ /**
+ * Apply some performance tuning
+ */
+ ProspectorUtils.addMRPerformance(job.getConfiguration());
+ }
+
+ job.setMapOutputKeyClass(IntermediateProspect.class);
+ job.setMapOutputValueClass(LongWritable.class);
+
+ job.setMapperClass(ProspectorMapper.class);
+ job.setCombinerClass(ProspectorCombiner.class);
+ job.setReducerClass(ProspectorReducer.class);
+ job.waitForCompletion(true);
+
+ final int success = job.isSuccessful() ? 0 : 1;
+
+ if (success == 0) {
+ final Mutation m = new Mutation(METADATA);
+ m.put(PROSPECT_TIME, getReverseIndexDateTime(truncatedDate), new ColumnVisibility(DEFAULT_VIS), truncatedDate.getTime(), new Value(EMPTY));
+ writeMutations(connector(instance(conf), conf), outTable, Collections.singleton(m));
+ }
+
+ return success;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorCombiner.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorCombiner.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorCombiner.java
new file mode 100644
index 0000000..bc3c1eb
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorCombiner.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.mr;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.rya.prospector.domain.IntermediateProspect;
+import org.apache.rya.prospector.plans.IndexWorkPlan;
+import org.apache.rya.prospector.plans.IndexWorkPlanManager;
+import org.apache.rya.prospector.plans.impl.ServicesBackedIndexWorkPlanManager;
+import org.apache.rya.prospector.utils.ProspectorUtils;
+
+/**
+ * Used to combine intermediate Prospect job results after {@link ProspectorMapper},
+ * but before the shuffle operation of the Hadoop Map Reduce framework.
+ */
+public class ProspectorCombiner extends Reducer<IntermediateProspect, LongWritable, IntermediateProspect, LongWritable> {
+
+ private final IndexWorkPlanManager manager = new ServicesBackedIndexWorkPlanManager();
+ private Map<String, IndexWorkPlan> plans;
+
+ @Override
+ public void setup(Context context) throws IOException, InterruptedException {
+ super.setup(context);
+ this.plans = ProspectorUtils.planMap(manager.getPlans());
+ }
+
+ @Override
+ protected void reduce(IntermediateProspect prospect, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
+ final IndexWorkPlan plan = plans.get(prospect.getIndex());
+ if (plan != null) {
+ final Collection<Entry<IntermediateProspect, LongWritable>> coll = plan.combine(prospect, values);
+ if (coll != null) {
+ for(final Entry<IntermediateProspect, LongWritable> entry : coll) {
+ context.write(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorMapper.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorMapper.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorMapper.java
new file mode 100644
index 0000000..ff4c30f
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorMapper.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.mr;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map.Entry;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.rya.accumulo.AccumuloRdfConfiguration;
+import org.apache.rya.api.RdfCloudTripleStoreConstants;
+import org.apache.rya.api.domain.RyaStatement;
+import org.apache.rya.api.resolver.RyaTripleContext;
+import org.apache.rya.api.resolver.triple.TripleRow;
+import org.apache.rya.api.resolver.triple.TripleRowResolverException;
+import org.apache.rya.prospector.domain.IntermediateProspect;
+import org.apache.rya.prospector.plans.IndexWorkPlan;
+import org.apache.rya.prospector.plans.IndexWorkPlanManager;
+import org.apache.rya.prospector.plans.impl.ServicesBackedIndexWorkPlanManager;
+
+/**
+ * Loads {@link RyaStatement}s from Accumulo and maps them into {@link IntermediateProspect}s
+ * paired with count information during the Map portion of the Hadoop Map Reduce framework.
+ */
+public class ProspectorMapper extends Mapper<Key, Value, IntermediateProspect, LongWritable> {
+
+ private RyaTripleContext ryaContext;
+ private final IndexWorkPlanManager manager = new ServicesBackedIndexWorkPlanManager();
+ private final Collection<IndexWorkPlan> plans = manager.getPlans();
+
+ @Override
+ public void setup(Context context) throws IOException, InterruptedException {
+ super.setup(context);
+ ryaContext = RyaTripleContext.getInstance(new AccumuloRdfConfiguration(context.getConfiguration()));
+ }
+
+ @Override
+ public void map(Key row, Value data, Context context) throws IOException, InterruptedException {
+ RyaStatement ryaStatement = null;
+ try {
+ ryaStatement = ryaContext.deserializeTriple(RdfCloudTripleStoreConstants.TABLE_LAYOUT.SPO,
+ new TripleRow(
+ row.getRow().getBytes(),
+ row.getColumnFamily().getBytes(),
+ row.getColumnQualifier().getBytes(),
+ row.getTimestamp(),
+ row.getColumnVisibility().getBytes(),
+ data.get()
+ )
+ );
+ } catch (final TripleRowResolverException e) {
+ // Do nothing. The row didn't contain a Rya Statement.
+ }
+
+ if(ryaStatement != null) {
+ for(final IndexWorkPlan plan : plans) {
+ final Collection<Entry<IntermediateProspect, LongWritable>> coll = plan.map(ryaStatement);
+ for(final Entry<IntermediateProspect, LongWritable> entry : coll) {
+ context.write(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorReducer.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorReducer.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorReducer.java
new file mode 100644
index 0000000..5247b5b
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/mr/ProspectorReducer.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.mr;
+
+import java.io.IOException;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.Map;
+
+import org.apache.commons.lang.time.DateUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.rya.prospector.domain.IntermediateProspect;
+import org.apache.rya.prospector.plans.IndexWorkPlan;
+import org.apache.rya.prospector.plans.IndexWorkPlanManager;
+import org.apache.rya.prospector.plans.impl.ServicesBackedIndexWorkPlanManager;
+import org.apache.rya.prospector.utils.ProspectorUtils;
+
+/**
+ * Reduces the {@link IntermediateProspect} counts into their final values and
+ * writes them to their final storage location during the Reduce step of the
+ * Hadoop Map Reduce framework.
+ */
+public class ProspectorReducer extends Reducer<IntermediateProspect, LongWritable, IntermediateProspect, LongWritable> {
+
+ private Date truncatedDate;
+ private final IndexWorkPlanManager manager = new ServicesBackedIndexWorkPlanManager();
+ private Map<String, IndexWorkPlan> plans;
+
+ @Override
+ public void setup(Context context) throws IOException, InterruptedException {
+ super.setup(context);
+
+ final Configuration conf = context.getConfiguration();
+ final long now = conf.getLong("DATE", System.currentTimeMillis());
+ truncatedDate = DateUtils.truncate(new Date(now), Calendar.MINUTE);
+
+ this.plans = ProspectorUtils.planMap(manager.getPlans());
+ }
+
+ @Override
+ protected void reduce(IntermediateProspect prospect, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
+ final IndexWorkPlan plan = plans.get(prospect.getIndex());
+ if (plan != null) {
+ plan.reduce(prospect, values, truncatedDate, context);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/IndexWorkPlan.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/IndexWorkPlan.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/IndexWorkPlan.java
new file mode 100644
index 0000000..77955e4
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/IndexWorkPlan.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.plans;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.rya.api.domain.RyaStatement;
+import org.apache.rya.prospector.domain.IndexEntry;
+import org.apache.rya.prospector.domain.IntermediateProspect;
+import org.apache.rya.prospector.mr.ProspectorCombiner;
+import org.apache.rya.prospector.mr.ProspectorMapper;
+import org.openrdf.model.vocabulary.XMLSchema;
+
+/**
+ * Contains the methods that perform each of the Map Reduce functions that result
+ * in the final {@link IndexEntry} values as well as a way to query those values
+ * once they have been written.
+ */
+public interface IndexWorkPlan {
+
+ public static final String URITYPE = XMLSchema.ANYURI.stringValue();
+ public static final LongWritable ONE = new LongWritable(1);
+ public static final String DELIM = "\u0000";
+
+ /**
+ * This method is invoked by {@link ProspectorMapper}. It's used to pull
+ * input from an Accumulo Rya instance into the Map Reduce framework.
+ * </p>
+ * It must use the values of a {@link RyaStatement} to derive a bunch of
+ * {@link IntermediateProspect} and {@code LongWritable} pairs. This is only
+ * useful for prospecting jobs that count things. The {@link IntermediateProspect}
+ * value will be used as the key within {@link #combine(IntermediateProspect, Iterable)} and
+ * {@link #reduce(IntermediateProspect, Iterable, Date, org.apache.hadoop.mapreduce.Reducer.Context)}.
+ *
+ * @param ryaStatement - The RDF Statement that needs to be mapped.
+ * @return A collection of intermediate keys and counts.
+ */
+ public Collection<Map.Entry<IntermediateProspect, LongWritable>> map(RyaStatement ryaStatement);
+
+ /**
+ * This method is invoked by {@link ProspectorCombiner}. It is used by to
+ * combine the results of {@link ProspectorMapper} before the shuffle operation
+ * of the Map Reduce framework.
+ *
+ * @param prospect - The intermediate prospect that is being combined.
+ * @param counts - The counts that need to be combined together.
+ * @return A collection containing the combined results.
+ */
+ public Collection<Map.Entry<IntermediateProspect, LongWritable>> combine(IntermediateProspect prospect, Iterable<LongWritable> counts);
+
+ /**
+ * This method is invoked by {@link ProsectorReducer}. It is used to reduce
+ * the counts to their final states and write them to output via the
+ * {@code context}.l
+ *
+ * @param prospect - The intermediate prospect that is being reduced.
+ * @param counts - The counts that need to be reduced.
+ * @param timestamp - The timestamp that identifies this Prospector run.
+ * @param context - The reducer context the reduced values will be written to.
+ * @throws IOException A problem was encountered while writing to the context.
+ * @throws InterruptedException Writes to the context were interrupted.
+ */
+ public void reduce(IntermediateProspect prospect, Iterable<LongWritable> counts, Date timestamp, Reducer.Context context) throws IOException, InterruptedException;
+
+ /**
+ * @return A unique name that indicates which {@link IndexEntry}s came from this plan.
+ */
+ public String getIndexType();
+
+ /**
+ * TODO Not sure what this generically is for. It is used by the count job to
+ * place a null delimiter between any {@link IndexEntry}s whose data
+ * section is two difference pieces of information together.
+ */
+ public String getCompositeValue(List<String> indices);
+
+ /**
+ * Search for {@link IndexEntry}s that have values matching the provided parameters.
+ *
+ * @param connector - The Accumulo Connector used to find the table holding the data.
+ * @param tableName - The name of the table the Prospector results are stored within.
+ * @param prospectTimes - Indicates which Prospect runs will be part of the query.
+ * @param type - The name of the index the {@link IndexEntry}s are stored within.
+ * @param index - The data portion of the {@link IndexEntry}s that may be returned.
+ * @param dataType - The data type of the {@link IndexEntry}s that may be returned.
+ * @param auths - The authorizations used to search for the entries.
+ * @return The {@link IndexEntries} that match the provided values.
+ * @throws TableNotFoundException No table exists for {@code tableName}.
+ */
+ public List<IndexEntry> query(Connector connector, String tableName, List<Long> prospectTimes, String type, String index, String dataType, String[] auths) throws TableNotFoundException;
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/IndexWorkPlanManager.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/IndexWorkPlanManager.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/IndexWorkPlanManager.java
new file mode 100644
index 0000000..1b7cf3b
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/IndexWorkPlanManager.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.plans;
+
+import java.util.Collection;
+
+import org.apache.rya.prospector.mr.Prospector;
+
+/**
+ * Provides access to the {@link IndexWorkPlan}s that will be executed as part
+ * of a {@link Prospector} run.
+ */
+public interface IndexWorkPlanManager {
+
+ /**
+ * @return The {@link IndexWorkPlan}s that will be executed as part of a
+ * {@link Prospector} run.
+ */
+ public Collection<IndexWorkPlan> getPlans();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/impl/CountPlan.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/impl/CountPlan.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/impl/CountPlan.java
new file mode 100644
index 0000000..ebcf6c3
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/impl/CountPlan.java
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.plans.impl;
+
+import static org.apache.rya.prospector.utils.ProspectorConstants.COUNT;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.rya.api.RdfCloudTripleStoreConstants;
+import org.apache.rya.api.domain.RyaStatement;
+import org.apache.rya.api.domain.RyaType;
+import org.apache.rya.api.domain.RyaURI;
+import org.apache.rya.prospector.domain.IndexEntry;
+import org.apache.rya.prospector.domain.IntermediateProspect;
+import org.apache.rya.prospector.domain.TripleValueType;
+import org.apache.rya.prospector.plans.IndexWorkPlan;
+import org.apache.rya.prospector.utils.CustomEntry;
+import org.apache.rya.prospector.utils.ProspectorUtils;
+import org.openrdf.model.util.URIUtil;
+import org.openrdf.model.vocabulary.XMLSchema;
+
+/**
+ * An implementation of {@link IndexWorkPlan} that counts the number of times
+ * a piece of data appears within a Rya Instance for every {@link TripleValueType}.
+ */
+public class CountPlan implements IndexWorkPlan {
+
+ @Override
+ public Collection<Map.Entry<IntermediateProspect, LongWritable>> map(RyaStatement ryaStatement) {
+ final RyaURI subject = ryaStatement.getSubject();
+ final RyaURI predicate = ryaStatement.getPredicate();
+ final String subjpred = ryaStatement.getSubject().getData() + DELIM + ryaStatement.getPredicate().getData();
+ final String predobj = ryaStatement.getPredicate().getData() + DELIM + ryaStatement.getObject().getData();
+ final String subjobj = ryaStatement.getSubject().getData() + DELIM + ryaStatement.getObject().getData();
+ final RyaType object = ryaStatement.getObject();
+ final int localIndex = URIUtil.getLocalNameIndex(subject.getData());
+ final String namespace = subject.getData().substring(0, localIndex - 1);
+ final String visibility = new String(ryaStatement.getColumnVisibility());
+
+ final List<Map.Entry<IntermediateProspect, LongWritable>> entries = new ArrayList<>(7);
+
+ // Create an entry for each TripleValueType type.
+ entries.add(new CustomEntry<IntermediateProspect, LongWritable>(
+ IntermediateProspect.builder()
+ .setIndex(COUNT)
+ .setData(subject.getData())
+ .setDataType(URITYPE)
+ .setTripleValueType( TripleValueType.SUBJECT )
+ .setVisibility(visibility)
+ .build()
+ , ONE));
+
+ entries.add(new CustomEntry<IntermediateProspect, LongWritable>(
+ IntermediateProspect.builder()
+ .setIndex(COUNT)
+ .setData(predicate.getData())
+ .setDataType(URITYPE)
+ .setTripleValueType( TripleValueType.PREDICATE )
+ .setVisibility(visibility)
+ .build()
+ , ONE));
+
+ entries.add(new CustomEntry<IntermediateProspect, LongWritable>(
+ IntermediateProspect.builder()
+ .setIndex(COUNT)
+ .setData(object.getData())
+ .setDataType(object.getDataType().stringValue())
+ .setTripleValueType( TripleValueType.OBJECT )
+ .setVisibility(visibility)
+ .build()
+ , ONE));
+
+ entries.add(new CustomEntry<IntermediateProspect, LongWritable>(
+ IntermediateProspect.builder()
+ .setIndex(COUNT)
+ .setData(subjpred)
+ .setDataType(XMLSchema.STRING.toString())
+ .setTripleValueType( TripleValueType.SUBJECT_PREDICATE )
+ .setVisibility(visibility)
+ .build()
+ , ONE));
+
+ entries.add(new CustomEntry<IntermediateProspect, LongWritable>(
+ IntermediateProspect.builder()
+ .setIndex(COUNT)
+ .setData(subjobj)
+ .setDataType(XMLSchema.STRING.toString())
+ .setTripleValueType(TripleValueType.SUBJECT_OBJECT)
+ .setVisibility(visibility)
+ .build()
+ , ONE));
+
+ entries.add(new CustomEntry<IntermediateProspect, LongWritable>(
+ IntermediateProspect.builder()
+ .setIndex(COUNT)
+ .setData(predobj)
+ .setDataType(XMLSchema.STRING.toString())
+ .setTripleValueType(TripleValueType.PREDICATE_OBJECT)
+ .setVisibility(visibility)
+ .build()
+ , ONE));
+
+ entries.add(new CustomEntry<IntermediateProspect, LongWritable>(
+ IntermediateProspect.builder()
+ .setIndex(COUNT)
+ .setData(namespace)
+ .setDataType(URITYPE)
+ .setTripleValueType(TripleValueType.ENTITY)
+ .setVisibility(visibility)
+ .build()
+ , ONE));
+ return entries;
+ }
+
+ @Override
+ public Collection<Map.Entry<IntermediateProspect, LongWritable>> combine(IntermediateProspect prospect, Iterable<LongWritable> counts) {
+ long sum = 0;
+ for(final LongWritable count : counts) {
+ sum += count.get();
+ }
+ return Collections.singleton( new CustomEntry<IntermediateProspect, LongWritable>(prospect, new LongWritable(sum)) );
+ }
+
+ @Override
+ public void reduce(IntermediateProspect prospect, Iterable<LongWritable> counts, Date timestamp, Reducer.Context context) throws IOException, InterruptedException {
+ long sum = 0;
+ for(final LongWritable count : counts) {
+ sum += count.get();
+ }
+
+ final String indexType = prospect.getTripleValueType().getIndexType();
+
+ // not sure if this is the best idea..
+ if ((sum >= 0) || indexType.equals(TripleValueType.PREDICATE.getIndexType())) {
+ final Mutation m = new Mutation(indexType + DELIM + prospect.getData() + DELIM + ProspectorUtils.getReverseIndexDateTime(timestamp));
+
+ final String dataType = prospect.getDataType();
+ final ColumnVisibility visibility = new ColumnVisibility(prospect.getVisibility());
+ final Value sumValue = new Value(("" + sum).getBytes());
+ m.put(COUNT, prospect.getDataType(), visibility, timestamp.getTime(), sumValue);
+
+ context.write(null, m);
+ }
+ }
+
+ @Override
+ public String getIndexType() {
+ return COUNT;
+ }
+
+ @Override
+ public String getCompositeValue(List<String> indices){
+ final Iterator<String> indexIt = indices.iterator();
+ String compositeIndex = indexIt.next();
+ while (indexIt.hasNext()){
+ final String value = indexIt.next();
+ compositeIndex += DELIM + value;
+ }
+ return compositeIndex;
+ }
+
+ @Override
+ public List<IndexEntry> query(Connector connector, String tableName, List<Long> prospectTimes, String type, String compositeIndex, String dataType, String[] auths) throws TableNotFoundException {
+ assert connector != null && tableName != null && type != null && compositeIndex != null;
+
+ final BatchScanner bs = connector.createBatchScanner(tableName, new Authorizations(auths), 4);
+ final List<Range> ranges = new ArrayList<>();
+ int max = 1000; //by default only return 1000 prospects maximum
+ if (prospectTimes != null) {
+ for(final Long prospectTime : prospectTimes) {
+ ranges.add(new Range(type + DELIM + compositeIndex + DELIM + ProspectorUtils.getReverseIndexDateTime(new Date(prospectTime))));
+ }
+ } else {
+ max = 1; //only return the latest if no prospectTimes given
+ final String prefix = type + DELIM + compositeIndex + DELIM;
+ ranges.add(new Range(prefix, prefix + RdfCloudTripleStoreConstants.LAST));
+ }
+
+ bs.setRanges(ranges);
+ if (dataType != null) {
+ bs.fetchColumn(new Text(COUNT), new Text(dataType));
+ } else {
+ bs.fetchColumnFamily(new Text(COUNT));
+ }
+
+ final List<IndexEntry> indexEntries = new ArrayList<IndexEntry>();
+ final Iterator<Entry<Key, Value>> iter = bs.iterator();
+
+ while (iter.hasNext() && indexEntries.size() <= max) {
+ final Entry<Key, Value> entry = iter.next();
+ final Key k = entry.getKey();
+ final Value v = entry.getValue();
+
+ final String[] rowArr = k.getRow().toString().split(DELIM);
+ String values = "";
+ // if it is a composite index, then return the type as a composite index
+ if (type.equalsIgnoreCase(TripleValueType.SUBJECT_PREDICATE.getIndexType()) ||
+ type.equalsIgnoreCase(TripleValueType.SUBJECT_OBJECT.getIndexType()) ||
+ type.equalsIgnoreCase(TripleValueType.PREDICATE_OBJECT.getIndexType())) {
+ values =rowArr[1] + DELIM + rowArr[2];
+ }
+ else {
+ values = rowArr[1];
+ }
+
+ // Create an entry using the values that were found.
+ final String entryDataType = k.getColumnQualifier().toString();
+ final String entryVisibility = k.getColumnVisibility().toString();
+ final Long entryCount = Long.parseLong(new String(v.get()));
+
+ indexEntries.add(
+ IndexEntry.builder()
+ .setData(values)
+ .setTripleValueType(rowArr[0])
+ .setIndex(COUNT)
+ .setDataType(entryDataType)
+ .setVisibility(entryVisibility)
+ .setCount(entryCount)
+ .setTimestamp(k.getTimestamp())
+ .build());
+ }
+ bs.close();
+
+ return indexEntries;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/impl/ServicesBackedIndexWorkPlanManager.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/impl/ServicesBackedIndexWorkPlanManager.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/impl/ServicesBackedIndexWorkPlanManager.java
new file mode 100644
index 0000000..b5d2320
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/plans/impl/ServicesBackedIndexWorkPlanManager.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.plans.impl;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+import org.apache.rya.prospector.plans.IndexWorkPlan;
+import org.apache.rya.prospector.plans.IndexWorkPlanManager;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Searches the classpath for any {@link IndexWorkPlan}s that are able to be service loaded.
+ */
+public class ServicesBackedIndexWorkPlanManager implements IndexWorkPlanManager {
+
+ private final Collection<IndexWorkPlan> plans;
+
+ /**
+ * Constructs an instance of {@link ServicesBackedIndexWorkPlanManager}.
+ */
+ public ServicesBackedIndexWorkPlanManager() {
+ final Iterator<IndexWorkPlan> iterator = ServiceLoader.load(IndexWorkPlan.class).iterator();
+ plans = Lists.newArrayList(iterator);
+ }
+
+ @Override
+ public Collection<IndexWorkPlan> getPlans() {
+ return plans;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/service/ProspectorService.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/service/ProspectorService.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/service/ProspectorService.java
new file mode 100644
index 0000000..205d4fc
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/service/ProspectorService.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.service;
+
+import static java.util.Objects.requireNonNull;
+import static org.apache.rya.prospector.utils.ProspectorConstants.METADATA;
+import static org.apache.rya.prospector.utils.ProspectorConstants.PROSPECT_TIME;
+
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.TableOperations;
+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.security.Authorizations;
+import org.apache.hadoop.io.Text;
+import org.apache.rya.prospector.domain.IndexEntry;
+import org.apache.rya.prospector.plans.IndexWorkPlan;
+import org.apache.rya.prospector.plans.IndexWorkPlanManager;
+import org.apache.rya.prospector.plans.impl.ServicesBackedIndexWorkPlanManager;
+import org.apache.rya.prospector.utils.ProspectorUtils;
+
+/**
+ * Provides access to the Prospect results that have been stored within a specific Accumulo table.
+ */
+public class ProspectorService {
+
+ private final Connector connector;
+ private final String tableName;
+
+ private final IndexWorkPlanManager manager = new ServicesBackedIndexWorkPlanManager();
+ private final Map<String, IndexWorkPlan> plans;
+
+ /**
+ * Constructs an instance of {@link ProspectorService}.
+ *
+ * @param connector - The Accumulo connector used to communicate with the table. (not null)
+ * @param tableName - The name of the Accumulo table that will be queried for Prospect results. (not null)
+ * @throws AccumuloException A problem occurred while creating the table.
+ * @throws AccumuloSecurityException A problem occurred while creating the table.
+ * @throws TableExistsException A problem occurred while creating the table.
+ */
+ public ProspectorService(Connector connector, String tableName) throws AccumuloException, AccumuloSecurityException, TableExistsException {
+ this.connector = requireNonNull(connector);
+ this.tableName = requireNonNull(tableName);
+
+ this.plans = ProspectorUtils.planMap(manager.getPlans());
+
+ // Create the table if it doesn't already exist.
+ final TableOperations tos = connector.tableOperations();
+ if(!tos.exists(tableName)) {
+ tos.create(tableName);
+ }
+ }
+
+ /**
+ * Get a list of timestamps that represents all of the Prospect runs that have
+ * ever been performed.
+ *
+ * @param auths - The authorizations used to scan the table for prospects.
+ * @return A list of timestamps representing each Prospect run that was found.
+ * @throws TableNotFoundException The table name that was provided when this
+ * class was constructed does not match a table that the connector has access to.
+ */
+ public Iterator<Long> getProspects(String[] auths) throws TableNotFoundException {
+ final Scanner scanner = connector.createScanner(tableName, new Authorizations(auths));
+ scanner.setRange(Range.exact(METADATA));
+ scanner.fetchColumnFamily(new Text(PROSPECT_TIME));
+
+ return new ProspectTimestampIterator( scanner.iterator() );
+ }
+
+ /**
+ * Get a list of timestamps that represents all of the Prospect runs that
+ * have been performed inclusively between two timestamps.
+ *
+ * @param beginTime - The start of the time range.
+ * @param endTime - The end of the time range.
+ * @param auths - The authorizations used to scan the table for prospects.
+ * @return A list of timestamps representing each Prospect run that was found.
+ * @throws TableNotFoundException The table name that was provided when this
+ * class was constructed does not match a table that the connector has access to.
+ */
+ public Iterator<Long> getProspectsInRange(long beginTime, long endTime, String[] auths) throws TableNotFoundException {
+ final Scanner scanner = connector.createScanner(tableName, new Authorizations(auths));
+ scanner.setRange(new Range(
+ new Key(METADATA, PROSPECT_TIME, ProspectorUtils.getReverseIndexDateTime(new Date(endTime)), "", Long.MAX_VALUE),
+ new Key(METADATA, PROSPECT_TIME, ProspectorUtils.getReverseIndexDateTime(new Date(beginTime)), "", 0l)
+ ));
+
+ return new ProspectTimestampIterator( scanner.iterator() );
+ }
+
+ /**
+ * Iterates over the results of a {@link Scanner} and interprets their keys
+ * contain Prospect run timestamps.
+ */
+ private static final class ProspectTimestampIterator implements Iterator<Long> {
+ private final Iterator<Entry<Key, Value>> it;
+
+ public ProspectTimestampIterator(Iterator<Entry<Key, Value>> it) {
+ this.it = requireNonNull(it);
+ }
+
+ @Override
+ public boolean hasNext() {
+ return it.hasNext();
+ }
+
+ @Override
+ public Long next() {
+ return it.next().getKey().getTimestamp();
+ }
+ }
+
+ /**
+ * Search for {@link IndexEntry}s that have values matching the provided parameters.
+ *
+ * @param prospectTimes - Indicates which Prospect runs will be part of the query.
+ * @param indexType - The name of the index the {@link IndexEntry}s are stored within.
+ * @param index - The data portion of the {@link IndexEntry}s that may be returned.
+ * @param dataType - The data type of the {@link IndexEntry}s that may be returned.
+ * @param auths - The authorizations used to search for the entries.
+ * @return The {@link IndexEntries} that match the provided values.
+ * @throws TableNotFoundException No table exists for {@code tableName}.
+ */
+ public List<IndexEntry> query(List<Long> prospectTimes, String indexType, String type, List<String> index, String dataType, String[] auths) throws TableNotFoundException {
+ assert indexType != null;
+
+ final IndexWorkPlan plan = plans.get(indexType);
+ assert plan != null: "Index Type: ${indexType} does not exist";
+ final String compositeIndex = plan.getCompositeValue(index);
+
+ return plan.query(connector, tableName, prospectTimes, type, compositeIndex, dataType, auths);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/service/ProspectorServiceEvalStatsDAO.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/service/ProspectorServiceEvalStatsDAO.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/service/ProspectorServiceEvalStatsDAO.java
new file mode 100644
index 0000000..3bb3b26
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/service/ProspectorServiceEvalStatsDAO.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.service;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.rya.api.RdfCloudTripleStoreConfiguration;
+import org.apache.rya.api.persist.RdfDAOException;
+import org.apache.rya.api.persist.RdfEvalStatsDAO;
+import org.apache.rya.prospector.domain.IndexEntry;
+import org.apache.rya.prospector.domain.TripleValueType;
+import org.apache.rya.prospector.utils.ProspectorConstants;
+import org.openrdf.model.Resource;
+import org.openrdf.model.Value;
+
+/**
+ * An ${@link org.apache.rya.api.persist.RdfEvalStatsDAO} that uses the Prospector Service underneath return counts.
+ */
+public class ProspectorServiceEvalStatsDAO implements RdfEvalStatsDAO<RdfCloudTripleStoreConfiguration> {
+
+ private ProspectorService prospectorService;
+
+ public ProspectorServiceEvalStatsDAO() {
+ }
+
+ public ProspectorServiceEvalStatsDAO(ProspectorService prospectorService, RdfCloudTripleStoreConfiguration conf) {
+ this.prospectorService = prospectorService;
+ }
+
+ public ProspectorServiceEvalStatsDAO(Connector connector, RdfCloudTripleStoreConfiguration conf) throws AccumuloException, AccumuloSecurityException, TableExistsException {
+ this.prospectorService = new ProspectorService(connector, getProspectTableName(conf));
+ }
+
+ @Override
+ public void init() {
+ assert prospectorService != null;
+ }
+
+ @Override
+ public boolean isInitialized() {
+ return prospectorService != null;
+ }
+
+ @Override
+ public void destroy() {
+ }
+
+ @Override
+ public double getCardinality(RdfCloudTripleStoreConfiguration conf, CARDINALITY_OF card, List<Value> val) throws RdfDAOException {
+ assert conf != null && card != null && val != null;
+
+ String triplePart = null;
+ switch (card) {
+ case SUBJECT:
+ triplePart = TripleValueType.SUBJECT.getIndexType();
+ break;
+ case PREDICATE:
+ triplePart = TripleValueType.PREDICATE.getIndexType();
+ break;
+ case OBJECT:
+ triplePart = TripleValueType.OBJECT.getIndexType();
+ break;
+ case SUBJECTPREDICATE:
+ triplePart = TripleValueType.SUBJECT_PREDICATE.getIndexType();
+ break;
+ case SUBJECTOBJECT:
+ triplePart = TripleValueType.SUBJECT_OBJECT.getIndexType();
+ break;
+ case PREDICATEOBJECT:
+ triplePart = TripleValueType.PREDICATE_OBJECT.getIndexType();
+ break;
+ }
+
+ final String[] auths = conf.getAuths();
+ final List<String> indexedValues = new ArrayList<>();
+ final Iterator<Value> valueIt = val.iterator();
+ while (valueIt.hasNext()){
+ indexedValues.add(valueIt.next().stringValue());
+ }
+
+ double cardinality = -1;
+ try {
+ final List<IndexEntry> entries = prospectorService.query(null, ProspectorConstants.COUNT, triplePart, indexedValues, null, auths);
+ if(!entries.isEmpty()) {
+ cardinality = entries.iterator().next().getCount();
+ }
+ } catch (final TableNotFoundException e) {
+ throw new RdfDAOException(e);
+ }
+ return cardinality;
+ }
+
+ @Override
+ public double getCardinality(RdfCloudTripleStoreConfiguration conf, CARDINALITY_OF card, List<Value> val, Resource context) {
+ return getCardinality(conf, card, val); //TODO: Not sure about the context yet
+ }
+
+ @Override
+ public void setConf(RdfCloudTripleStoreConfiguration conf) {
+ }
+
+ @Override
+ public RdfCloudTripleStoreConfiguration getConf() {
+ return null;
+ }
+
+ public static String getProspectTableName(RdfCloudTripleStoreConfiguration conf) {
+ return conf.getTablePrefix() + "prospects";
+ }
+
+ /**
+ * This method exists so that the Rya Web project may autowrire itself together
+ * using the Spring framework.
+ *
+ * @param prospectorService - The {@link ProspectorService} that will be used by this DAO.
+ */
+ public void setProspectorService(ProspectorService prospectorService) {
+ this.prospectorService = prospectorService;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/CustomEntry.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/CustomEntry.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/CustomEntry.java
new file mode 100644
index 0000000..bff2c42
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/CustomEntry.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.utils;
+
+import java.util.Map;
+
+/**
+ * A convenience class that implements {@link Map.Entry}.
+ *
+ * @param <K> - The type of the Key.
+ * @param <V> - The type of the Value.
+ */
+public class CustomEntry<K, V> implements Map.Entry<K, V> {
+
+ private K key;
+ private V value;
+
+ public CustomEntry(K key, V value) {
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ public void setKey(K key) {
+ this.key = key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ this.value = value;
+ return this.value;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/ProspectorConstants.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/ProspectorConstants.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/ProspectorConstants.java
new file mode 100644
index 0000000..7901343
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/ProspectorConstants.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.utils;
+
+/**
+ * Constants used by the Prospector project.
+ */
+public class ProspectorConstants {
+ /**
+ * The name of the Count index.
+ */
+ public static final String COUNT = "count";
+
+ /**
+ * The Row ID within Accumulo for any metadata entry related to a Prospect run.
+ */
+ public static final String METADATA = "metadata";
+
+ /**
+ * This is the name of a Column Family within Accumulo that represents when
+ * a Prospect run was performed.
+ */
+ public static final String PROSPECT_TIME = "prospectTime";
+
+ public static final String DEFAULT_VIS = "U&FOUO";
+ public static final byte[] EMPTY = new byte [0];
+
+ //config properties
+ public static final String PERFORMANT = "performant";
+
+ public static final String USERNAME = "username";
+ public static final String PASSWORD = "password";
+ public static final String INSTANCE = "instance";
+ public static final String ZOOKEEPERS = "zookeepers";
+ public static final String MOCK = "mock";
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/ProspectorUtils.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/ProspectorUtils.java b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/ProspectorUtils.java
new file mode 100644
index 0000000..4dc9253
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/org/apache/rya/prospector/utils/ProspectorUtils.java
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.rya.prospector.utils;
+
+import static org.apache.rya.prospector.utils.ProspectorConstants.INSTANCE;
+import static org.apache.rya.prospector.utils.ProspectorConstants.MOCK;
+import static org.apache.rya.prospector.utils.ProspectorConstants.PASSWORD;
+import static org.apache.rya.prospector.utils.ProspectorConstants.USERNAME;
+import static org.apache.rya.prospector.utils.ProspectorConstants.ZOOKEEPERS;
+
+import java.text.SimpleDateFormat;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
+import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
+import org.apache.accumulo.core.client.mock.MockInstance;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.commons.lang.Validate;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.compress.GzipCodec;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.rya.prospector.plans.IndexWorkPlan;
+
+public class ProspectorUtils {
+
+ public static final long INDEXED_DATE_SORT_VAL = 999999999999999999L; // 18 char long, same length as date format pattern below
+ public static final String INDEXED_DATE_FORMAT = "yyyyMMddHHmmsssSSS";
+
+ public static String getReverseIndexDateTime(Date date) {
+ Validate.notNull(date);
+ final String formattedDateString = new SimpleDateFormat(INDEXED_DATE_FORMAT).format(date);
+ final long diff = INDEXED_DATE_SORT_VAL - Long.valueOf(formattedDateString);
+
+ return Long.toString(diff);
+ }
+
+ public static Map<String, IndexWorkPlan> planMap(Collection<IndexWorkPlan> plans) {
+ final Map<String, IndexWorkPlan> planMap = new HashMap<>();
+ for(final IndexWorkPlan plan : plans) {
+ planMap.put(plan.getIndexType(), plan);
+ }
+ return planMap;
+ }
+
+ public static void initMRJob(Job job, String table, String outtable, String[] auths) throws AccumuloSecurityException {
+ final Configuration conf = job.getConfiguration();
+ final String username = conf.get(USERNAME);
+ final String password = conf.get(PASSWORD);
+ final String instance = conf.get(INSTANCE);
+ final String zookeepers = conf.get(ZOOKEEPERS);
+ final String mock = conf.get(MOCK);
+
+ //input
+ if (Boolean.parseBoolean(mock)) {
+ AccumuloInputFormat.setMockInstance(job, instance);
+ AccumuloOutputFormat.setMockInstance(job, instance);
+ } else if (zookeepers != null) {
+ AccumuloInputFormat.setZooKeeperInstance(job, instance, zookeepers);
+ AccumuloOutputFormat.setZooKeeperInstance(job, instance, zookeepers);
+ } else {
+ throw new IllegalArgumentException("Must specify either mock or zookeepers");
+ }
+
+ AccumuloInputFormat.setConnectorInfo(job, username, new PasswordToken(password.getBytes()));
+ AccumuloInputFormat.setInputTableName(job, table);
+ job.setInputFormatClass(AccumuloInputFormat.class);
+ AccumuloInputFormat.setScanAuthorizations(job, new Authorizations(auths));
+
+ // OUTPUT
+ job.setOutputFormatClass(AccumuloOutputFormat.class);
+ job.setOutputKeyClass(Text.class);
+ job.setOutputValueClass(Mutation.class);
+ AccumuloOutputFormat.setConnectorInfo(job, username, new PasswordToken(password.getBytes()));
+ AccumuloOutputFormat.setDefaultTableName(job, outtable);
+ }
+
+ public static void addMRPerformance(Configuration conf) {
+ conf.setBoolean("mapred.map.tasks.speculative.execution", false);
+ conf.setBoolean("mapred.reduce.tasks.speculative.execution", false);
+ conf.set("io.sort.mb", "256");
+ conf.setBoolean("mapred.compress.map.output", true);
+ conf.set("mapred.map.output.compression.codec", GzipCodec.class.getName());
+ }
+
+ public static Instance instance(Configuration conf) {
+ assert conf != null;
+
+ final String instance_str = conf.get(INSTANCE);
+ final String zookeepers = conf.get(ZOOKEEPERS);
+ final String mock = conf.get(MOCK);
+ if (Boolean.parseBoolean(mock)) {
+ return new MockInstance(instance_str);
+ } else if (zookeepers != null) {
+ return new ZooKeeperInstance(instance_str, zookeepers);
+ } else {
+ throw new IllegalArgumentException("Must specify either mock or zookeepers");
+ }
+ }
+
+ public static Connector connector(Instance instance, Configuration conf) throws AccumuloException, AccumuloSecurityException {
+ final String username = conf.get(USERNAME);
+ final String password = conf.get(PASSWORD);
+ if (instance == null) {
+ instance = instance(conf);
+ }
+ return instance.getConnector(username, new PasswordToken(password));
+ }
+
+ public static void writeMutations(Connector connector, String tableName, Collection<Mutation> mutations) throws TableNotFoundException, MutationsRejectedException {
+ final BatchWriter bw = connector.createBatchWriter(tableName, 10000l, 10000l, 4);
+ for(final Mutation mutation : mutations) {
+ bw.addMutation(mutation);
+ }
+ bw.flush();
+ bw.close();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/bd9b9124/extras/rya.prospector/src/test/groovy/org/apache/rya/prospector/mr/ProspectorTest.groovy
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/groovy/org/apache/rya/prospector/mr/ProspectorTest.groovy b/extras/rya.prospector/src/test/groovy/org/apache/rya/prospector/mr/ProspectorTest.groovy
deleted file mode 100644
index 02cbcd1..0000000
--- a/extras/rya.prospector/src/test/groovy/org/apache/rya/prospector/mr/ProspectorTest.groovy
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.rya.prospector.mr
-
-import com.google.common.collect.Iterators
-import com.google.common.collect.Lists
-import org.apache.rya.accumulo.AccumuloRyaDAO
-import org.apache.rya.accumulo.AccumuloRdfConfiguration
-import org.apache.rya.api.persist.RdfEvalStatsDAO
-import org.apache.rya.api.domain.RyaStatement
-import org.apache.rya.api.domain.RyaType
-import org.apache.rya.api.domain.RyaURI
-import org.apache.rya.prospector.domain.IndexEntry
-import org.apache.rya.prospector.domain.TripleValueType
-import org.apache.rya.prospector.service.ProspectorService
-import org.apache.rya.prospector.service.ProspectorServiceEvalStatsDAO
-import org.apache.rya.prospector.utils.ProspectorConstants
-import org.apache.accumulo.core.client.Instance
-import org.apache.accumulo.core.client.mock.MockInstance
-import org.apache.accumulo.core.security.Authorizations
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
-import org.apache.hadoop.util.ToolRunner
-import org.junit.Test
-import org.openrdf.model.vocabulary.XMLSchema
-import org.openrdf.model.impl.URIImpl
-
-import static org.junit.Assert.assertEquals
-import org.openrdf.model.impl.LiteralImpl
-import org.openrdf.model.Value
-
-/**
- * Date: 12/4/12
- * Time: 4:33 PM
- */
-class ProspectorTest {
-
- @Test
- public void testCount() throws Exception {
-
- Instance mock = new MockInstance("accumulo");
-
- def connector = mock.getConnector("user", "pass".bytes)
- def intable = "rya_spo"
- def outtable = "rya_prospects"
- if (connector.tableOperations().exists(outtable))
- connector.tableOperations().delete(outtable)
- connector.tableOperations().create(outtable)
-
- AccumuloRyaDAO ryaDAO = new AccumuloRyaDAO();
- ryaDAO.setConnector(connector);
- ryaDAO.init()
-
- ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("mydata1")))
- ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("mydata2")))
- ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("12")))
- ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1235"), new RyaURI("urn:gem#pred"), new RyaType(XMLSchema.INTEGER, "12")))
- ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1235"), new RyaURI("urn:gem#pred1"), new RyaType("12")))
-
- def confFile = "stats_cluster_config.xml"
- def confPath = new Path(getClass().getClassLoader().getResource(confFile).toString())
- def args = (String[]) [confPath];
- ToolRunner.run(new Prospector(), args);
- debugTable(connector, outtable)
-
- def scanner = connector.createScanner(outtable, new Authorizations("U", "FOUO"))
- def iter = scanner.iterator()
-// assertEquals(11, Iterators.size(iter))
-
- ryaDAO.destroy()
-
- def conf = new Configuration()
- conf.addResource(confPath)
- // debugTable(mrInfo, outtable)
-
- def service = new ProspectorService(connector, outtable)
- def auths = (String[]) ["U", "FOUO"]
- def prospects = service.getProspects(auths)
- def plist = Lists.newArrayList(prospects)
- assertEquals(1, plist.size())
-
- def rdfConf = new AccumuloRdfConfiguration(conf)
- rdfConf.setAuths("U","FOUO")
-
- prospects = service.getProspectsInRange(System.currentTimeMillis() - 100000, System.currentTimeMillis() + 10000, auths)
- plist = Lists.newArrayList(prospects)
- assertEquals(1, plist.size())
-
- List<String> queryTerms = new ArrayList<String>();
- queryTerms.add("urn:gem:etype");
- def query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.entity.name(), queryTerms, XMLSchema.ANYURI.stringValue(), auths)
- assertEquals(1, query.size())
-// assertEquals(
-// new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem:etype", dataType: XMLSchema.ANYURI.stringValue(),
-// tripleValueType: TripleValueType.entity, visibility: "", count: -1, timestamp: plist.get(0)),
-// query.get(0))
-
- queryTerms = new ArrayList<String>();
- queryTerms.add("urn:gem:etype#1234");
- query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.subject.name(), queryTerms, XMLSchema.ANYURI.stringValue(), auths)
- assertEquals(1, query.size())
-
- queryTerms = new ArrayList<String>();
- queryTerms.add("urn:gem#pred");
- query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.predicate.name(), queryTerms, XMLSchema.ANYURI.stringValue(), auths)
- assertEquals(1, query.size())
- assertEquals(
- new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem#pred", dataType: XMLSchema.ANYURI.stringValue(),
- tripleValueType: TripleValueType.predicate, visibility: "", count: 4l, timestamp: plist.get(0)),
- query.get(0))
-
- queryTerms = new ArrayList<String>();
- queryTerms.add("mydata1");
- query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.object.name(), queryTerms, XMLSchema.STRING.stringValue(), auths)
- assertEquals(1, query.size())
-// assertEquals(
-// new IndexEntry(index: ProspectorConstants.COUNT, data: "mydata1", dataType: XMLSchema.STRING.stringValue(),
-// tripleValueType: TripleValueType.object, visibility: "", count: -1, timestamp: plist.get(0)),
-// query.get(0))
-
- queryTerms = new ArrayList<String>();
- queryTerms.add("urn:gem:etype#1234");
- queryTerms.add("urn:gem#pred");
- query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.subjectpredicate.name(), queryTerms, XMLSchema.STRING.stringValue(), auths)
- assertEquals(1, query.size())
-// assertEquals(
-// new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem:etype#1234" + "\u0000" + "urn:gem#pred", dataType: XMLSchema.STRING.stringValue(),
-// tripleValueType: TripleValueType.subjectpredicate, visibility: "", count: -1, timestamp: plist.get(0)),
-// query.get(0))
-
- queryTerms = new ArrayList<String>();
- queryTerms.add("urn:gem#pred");
- queryTerms.add("12");
- query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.predicateobject.name(), queryTerms, XMLSchema.STRING.stringValue(), auths)
- assertEquals(1, query.size())
-// assertEquals(
-// new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem#pred" + "\u0000" + "12", dataType: XMLSchema.STRING.stringValue(),
-// tripleValueType: TripleValueType.predicateobject, visibility: "", count: -1, timestamp: plist.get(0)),
-// query.get(0))
-
- queryTerms = new ArrayList<String>();
- queryTerms.add("urn:gem:etype#1234");
- queryTerms.add("mydata1");
- query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.subjectobject.name(), queryTerms, XMLSchema.STRING.stringValue(), auths)
-
- assertEquals(1, query.size())
-// assertEquals(
-// new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem:etype#1234" + "\u0000" + "mydata1", dataType: XMLSchema.STRING.stringValue(),
-// tripleValueType: TripleValueType.subjectobject, visibility: "", count: -1, timestamp: plist.get(0)),
-// query.get(0))
-
- //should be in a teardown method
- connector.tableOperations().delete(outtable)
- }
-
- private void debugTable(def connector, String table) {
- connector.createScanner(table, new Authorizations((String[]) ["U", "FOUO"])).iterator().each {
- println it
- }
- }
-}