Prospector extends Configured implements Tool { @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)) { 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; }  static void main(String[] args); @Override int run(String[] args);  }
@Test public void testCount() throws Exception { final Instance mock = new MockInstance("accumulo"); final Connector connector = mock.getConnector("user", new PasswordToken("pass")); final String outtable = "rya_prospects"; if (connector.tableOperations().exists(outtable)) { connector.tableOperations().delete(outtable); } connector.tableOperations().create(outtable); final AccumuloRyaDAO ryaDAO = new AccumuloRyaDAO(); ryaDAO.setConnector(connector); ryaDAO.init(); ryaDAO.add(new RyaStatement(new RyaIRI("urn:gem:etype#1234"), new RyaIRI("urn:gem#pred"), new RyaType("mydata1"))); ryaDAO.add(new RyaStatement(new RyaIRI("urn:gem:etype#1234"), new RyaIRI("urn:gem#pred"), new RyaType("mydata2"))); ryaDAO.add(new RyaStatement(new RyaIRI("urn:gem:etype#1234"), new RyaIRI("urn:gem#pred"), new RyaType("12"))); ryaDAO.add(new RyaStatement(new RyaIRI("urn:gem:etype#1235"), new RyaIRI("urn:gem#pred"), new RyaType(XMLSchema.INTEGER, "12"))); ryaDAO.add(new RyaStatement(new RyaIRI("urn:gem:etype#1235"), new RyaIRI("urn:gem#pred1"), new RyaType("12"))); final String confFile = "stats_cluster_config.xml"; final Path confPath = new Path(getClass().getClassLoader().getResource(confFile).toString()); final String[] args = { confPath.toString() }; ToolRunner.run(new Prospector(), args); ryaDAO.destroy(); final Configuration conf = new Configuration(); conf.addResource(confPath); final ProspectorService service = new ProspectorService(connector, outtable); final String[] auths = {"U", "FOUO"}; Iterator<Long> prospects = service.getProspects(auths); List<Long> plist = Lists.newArrayList(prospects); assertEquals(1, plist.size()); final Long prospectTimestamp = plist.iterator().next(); final AccumuloRdfConfiguration 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<>(); queryTerms.add("urn:gem:etype"); final List<IndexEntry> entityEntries = service.query(plist, ProspectorConstants.COUNT, TripleValueType.ENTITY.getIndexType(), queryTerms, XMLSchema.ANYURI.stringValue(), auths); final List<IndexEntry> expectedEntityEntries = Lists.newArrayList( IndexEntry.builder() .setIndex(ProspectorConstants.COUNT) .setData("urn:gem:etype") .setDataType(XMLSchema.ANYURI.stringValue()) .setTripleValueType( TripleValueType.ENTITY.getIndexType() ) .setVisibility("") .setTimestamp(prospectTimestamp) .setCount(new Long(5)) .build()); assertEquals(expectedEntityEntries, entityEntries); queryTerms = new ArrayList<String>(); queryTerms.add("urn:gem:etype#1234"); final List<IndexEntry> subjectEntries = service.query(plist, ProspectorConstants.COUNT, TripleValueType.SUBJECT.getIndexType(), queryTerms, XMLSchema.ANYURI.stringValue(), auths); final List<IndexEntry> expectedSubjectEntries = Lists.newArrayList( IndexEntry.builder() .setIndex(ProspectorConstants.COUNT) .setData("urn:gem:etype#1234") .setDataType(XMLSchema.ANYURI.stringValue()) .setTripleValueType( TripleValueType.SUBJECT.getIndexType() ) .setVisibility("") .setTimestamp(prospectTimestamp) .setCount(new Long(3)) .build()); assertEquals(expectedSubjectEntries, subjectEntries); queryTerms = new ArrayList<String>(); queryTerms.add("urn:gem#pred"); final List<IndexEntry> predicateEntries = service.query(plist, ProspectorConstants.COUNT, TripleValueType.PREDICATE.getIndexType(), queryTerms, XMLSchema.ANYURI.stringValue(), auths); final List<IndexEntry> expectedPredicateEntries = Lists.newArrayList( IndexEntry.builder() .setIndex(ProspectorConstants.COUNT) .setData("urn:gem#pred") .setDataType(XMLSchema.ANYURI.stringValue()) .setTripleValueType( TripleValueType.PREDICATE.getIndexType() ) .setVisibility("") .setTimestamp(prospectTimestamp) .setCount(new Long(4)) .build()); assertEquals(expectedPredicateEntries, predicateEntries); queryTerms = new ArrayList<String>(); queryTerms.add("mydata1"); final List<IndexEntry> objectEntries = service.query(plist, ProspectorConstants.COUNT, TripleValueType.OBJECT.getIndexType(), queryTerms, XMLSchema.STRING.stringValue(), auths); final List<IndexEntry> expectedObjectEntries = Lists.newArrayList( IndexEntry.builder() .setIndex(ProspectorConstants.COUNT) .setData("mydata1") .setDataType(XMLSchema.STRING.stringValue()) .setTripleValueType( TripleValueType.OBJECT.getIndexType() ) .setVisibility("") .setTimestamp(prospectTimestamp) .setCount(new Long(1)) .build()); assertEquals(expectedObjectEntries, objectEntries); queryTerms = new ArrayList<String>(); queryTerms.add("urn:gem:etype#1234"); queryTerms.add("urn:gem#pred"); final List<IndexEntry> subjectPredicateEntries = service.query(plist, ProspectorConstants.COUNT, TripleValueType.SUBJECT_PREDICATE.getIndexType(), queryTerms, XMLSchema.STRING.stringValue(), auths); final List<IndexEntry> expectedSubjectPredicateEntries = Lists.newArrayList( IndexEntry.builder() .setIndex(ProspectorConstants.COUNT) .setData("urn:gem:etype#1234"+ "\u0000" + "urn:gem#pred") .setDataType(XMLSchema.STRING.stringValue()) .setTripleValueType( TripleValueType.SUBJECT_PREDICATE.getIndexType() ) .setVisibility("") .setTimestamp(prospectTimestamp) .setCount(new Long(3)) .build()); assertEquals(expectedSubjectPredicateEntries, subjectPredicateEntries); queryTerms = new ArrayList<String>(); queryTerms.add("urn:gem#pred"); queryTerms.add("12"); final List<IndexEntry> predicateObjectEntries = service.query(plist, ProspectorConstants.COUNT, TripleValueType.PREDICATE_OBJECT.getIndexType(), queryTerms, XMLSchema.STRING.stringValue(), auths); final List<IndexEntry> expectedPredicateObjectEntries = Lists.newArrayList( IndexEntry.builder() .setIndex(ProspectorConstants.COUNT) .setData("urn:gem#pred" + "\u0000" + "12") .setDataType(XMLSchema.STRING.stringValue()) .setTripleValueType( TripleValueType.PREDICATE_OBJECT.getIndexType() ) .setVisibility("") .setTimestamp(prospectTimestamp) .setCount(new Long(2)) .build()); assertEquals(expectedPredicateObjectEntries, predicateObjectEntries); queryTerms = new ArrayList<String>(); queryTerms.add("urn:gem:etype#1234"); queryTerms.add("mydata1"); final List<IndexEntry> subjectObjectEntries = service.query(plist, ProspectorConstants.COUNT, TripleValueType.SUBJECT_OBJECT.getIndexType(), queryTerms, XMLSchema.STRING.stringValue(), auths); final List<IndexEntry> expectedSubjectObjectEntries = Lists.newArrayList( IndexEntry.builder() .setIndex(ProspectorConstants.COUNT) .setData("urn:gem:etype#1234" + "\u0000" + "mydata1") .setDataType(XMLSchema.STRING.stringValue()) .setTripleValueType( TripleValueType.SUBJECT_OBJECT.getIndexType() ) .setVisibility("") .setTimestamp(prospectTimestamp) .setCount(new Long(1)) .build()); assertEquals(expectedSubjectObjectEntries, subjectObjectEntries); }