| /* |
| * 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.nio.charset.StandardCharsets; |
| 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.eclipse.rdf4j.model.util.URIUtil; |
| import org.eclipse.rdf4j.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(final 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(), StandardCharsets.UTF_8); |
| |
| 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(final IntermediateProspect prospect, final 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(final IntermediateProspect prospect, final Iterable<LongWritable> counts, final Date timestamp, final 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(StandardCharsets.UTF_8)); |
| m.put(COUNT, prospect.getDataType(), visibility, timestamp.getTime(), sumValue); |
| |
| context.write(null, m); |
| } |
| } |
| |
| @Override |
| public String getIndexType() { |
| return COUNT; |
| } |
| |
| @Override |
| public String getCompositeValue(final 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(final Connector connector, final String tableName, final List<Long> prospectTimes, final String type, final String compositeIndex, final String dataType, final 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(), StandardCharsets.UTF_8)); |
| |
| 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; |
| } |
| } |
